{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Improving Predictions\n",
    "\n",
    "Now that we have deployed working models predicting flight delays, it is time to “make believe” that our prediction has proven useful based on user feedback, and further that the prediction is valuable enough that prediction quality is important. In this case, it is time to iteratively improve the quality of our prediction. If a prediction is valuable enough, this becomes a full-time job for one or more people.\n",
    "\n",
    "In this chapter we will tune our Spark ML classifier and also do additional feature engineering to improve prediction quality. In doing so, we will show you how to iteratively improve predictions.\n",
    "\n",
    "## Fixing Our Prediction Problem\n",
    "\n",
    "At this point we realized that our model was always predicting one class, no matter the input. We began by investigating that in a Jupyter notebook at [ch09/Debugging Prediction Problems.ipynb](Debugging Prediction Problems.ipynb).\n",
    "\n",
    "The notebook itself is very long, and we tried many things to fix our model. It turned out we had made a mistake. We were using `OneHotEncoder` on top of the output of `StringIndexerModel` when we were encoding our nominal/categorical string features. This is how you should encode features for models other than decision trees, but it turns out that for decision tree models, you are supposed to take the string indexes from `StringIndexerModel` and directly compose them with your continuous/numeric features in a `VectorAssembler`. Decision trees are able to infer the fact that indexes represent categories. One benefit of directly adding StringIndexes to your feature vectors is that you then get easily interpretable feature importances.\n",
    "\n",
    "When we discovered this, we had to go back and edit the book so that we didn’t teach something that was wrong, and so this is now what you see. We thought it worthwhile to link to the notebook, though, to show how this really works in the wild: you build broken shit and then fix it.\n",
    "\n",
    "## When to Improve Predictions\n",
    "\n",
    "Not all predictions should be improved. Often something fast and crude will work well enough as an MVP (minimum viable product). Only predictions that prove useful should be improved. It is possible to sink large volumes of time into improving the quality of a prediction, so it is essential that you connect with users before getting sucked into this task. This is why we’ve included the discussion of improving predictions in its own chapter.\n",
    "\n",
    "## Improving Prediction Performance\n",
    "\n",
    "There are a few ways to improve an existing predictive model. The first is by tuning the parameters of the statistical model making your prediction. The second is feature engineering.\n",
    "\n",
    "Tuning model hyperparameters to improve predictive model quality can be done by intuition, or by brute force through something called a grid or random search. We’re going to focus on feature engineering, as hyperparameter tuning is covered elsewhere. A good guide to hyperparameter tuning is available in the Spark documentation on model selection and tuning.\n",
    "\n",
    "As we move through this chapter, we’ll be using the work we’ve done so far to perform feature engineering. Feature engineering is the most important part of making good predictions. It involves using what you’ve discovered about the data through exploratory data analysis in order to feed your machine learning algorithm better, more consequential data as input.\n",
    "\n",
    "### Experimental Adhesion Method: See What Sticks\n",
    "\n",
    "There are several ways to decide which features to use, and Saurav Kaushik has written a post on Analytics Vidhya that introduces them well. The method we employ primarily, which we jokingly entitle the Experimental Adhesion Method, is to quickly select all the features that we can simply compute, and try them all using a random forest or gradient boosted decision tree model (note that even if our application requires another type of model, we still use decision trees to guide feature selection). Then we train the model and inspect the model’s feature importances to “see what sticks.” The most important variables are retained, and this forms the basic model we begin with.\n",
    "\n",
    "Feature engineering is an iterative process. Based on the feature importances, we ponder what new things we might try using the data we have available. We start with the simplest idea, or the one that is easiest to implement. If the feature importances indicate one type of feature is important, and we can’t easily compute new features similar to this one, we think about how we might acquire new data to join to our training data to use as features.\n",
    "\n",
    "The key is to be logical and systematic in our exploration of the feature space. You should think about how easy a potential feature is to compute, as well as what it would teach you if it turned out to be important. Are there other, similar features that you could try if this candidate worked? Develop hypotheses and test them in the form of new features. Evaluate each new feature in an experiment and reflect on what you’ve learned before engineering the next feature.\n",
    "\n",
    "### Establishing Rigorous Metrics for Experiments\n",
    "\n",
    "In order to improve our classification model, we need to reliably determine its prediction quality in the first place. To do so, we need to beef up our cross-validation code, and then establish a baseline of quality for the original model. Check out [ch09/baseline_spark_mllib_model.py](baseline_spark_mllib_model.py), which we copied from [ch09/train_spark_mllib_model.py](train_spark_mllib_model.py) and altered to improve its cross-validation code.\n",
    "\n",
    "In order to evaluate the prediction quality of our classifier, we need to use more than one metric. Spark ML’s `MulticlassClassificationEvaluator` offers four metrics: accuracy, weighted precision, weighted recall, and f1.\n",
    "\n",
    "### Defining Our Classification Metrics\n",
    "\n",
    "The raw _accuracy_ is just what it sounds like: the number of correct predictions divided by the number of predictions. This is something to check first, but it isn’t adequate alone. _Precision_ is a measure of how useful the result is. _Recall_ describes how complete the results are. The _f1_ score incorporates both precision and recall to determine overall quality. Taken together, the changes to these metrics between consecutive runs of training our model can give us a clear picture of what is happening with our model in terms of prediction quality. We will use these metrics along with feature importance to guide our feature engineering efforts.\n",
    "\n",
    "### Feature Importance\n",
    "\n",
    "Model quality metrics aren’t enough to guide the iterative improvements of our model. To understand what is going on with each new run, we need to employ a type of model called a decision tree.\n",
    "\n",
    "In Spark ML, the best general-purpose multiclass classification model is an implementation of a random forest, the RandomForestClassificationModel, fit by the RandomForestClassifier. Random forests can classify or regress, and they have an important feature that helps us interrogate predictive models through a feature called feature importance.\n",
    "\n",
    "The importance of a feature is what it sounds like: a measure of how important that feature was in contributing to the accuracy of the model. This information is incredibly useful, as it can serve as a guiding hand to feature engineering. In other words, if you know how important a feature is, you can use this clue to make changes that increase the accuracy of the model, such as removing unimportant features and trying to engineer features similar to those that are most important. Feature engineering is a major theme of Agile Data Science, and it is a big part of why we’ve been doing iterative visualization and exploration (the purpose of which is to shed light on and drive feature engineering).\n",
    "\n",
    "Note that the state of the art for many classification and regression tasks is a gradient boosted decision tree, but as of version 2.1.0 Spark ML’s implementation—the `GBTClassificationModel`, which is fit by the `GBTClassifier`—can only do binary classification.\n",
    "\n",
    "### Getting Ready for Experiments\n",
    "\n",
    "We need to run through the model's code from chapter 8 before we can setup and run an experiment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PySpark initialized...\n"
     ]
    }
   ],
   "source": [
    "import sys, os, re\n",
    "import json\n",
    "import datetime, iso8601\n",
    "from tabulate import tabulate\n",
    "\n",
    "# Initialize PySpark\n",
    "APP_NAME = \"Improving Predictions\"\n",
    "\n",
    "# If there is no SparkSession, create the environment\n",
    "try:\n",
    "    sc and spark\n",
    "except NameError as e:\n",
    "    import findspark\n",
    "    findspark.init()\n",
    "    import pyspark\n",
    "    import pyspark.sql\n",
    "\n",
    "    sc = pyspark.SparkContext()\n",
    "    spark = pyspark.sql.SparkSession(sc).builder.appName(APP_NAME).getOrCreate()\n",
    "\n",
    "print(\"PySpark initialized...\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "Row(ArrDelay=43.0, CRSArrTime=datetime.datetime(2015, 1, 2, 22, 55), CRSDepTime=datetime.datetime(2015, 1, 2, 22, 5), Carrier='WN', DayOfMonth=2, DayOfWeek=5, DayOfYear=2, DepDelay=35.0, Dest='CRP', Distance=187.0, FlightDate=datetime.date(2015, 1, 2), FlightNum='1981', Origin='HOU')"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType, DateType, TimestampType\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "schema = StructType([\n",
    "    StructField(\"ArrDelay\", DoubleType(), True),     # \"ArrDelay\":5.0\n",
    "    StructField(\"CRSArrTime\", TimestampType(), True),    # \"CRSArrTime\":\"2015-12-31T03:20:00.000-08:00\"\n",
    "    StructField(\"CRSDepTime\", TimestampType(), True),    # \"CRSDepTime\":\"2015-12-31T03:05:00.000-08:00\"\n",
    "    StructField(\"Carrier\", StringType(), True),     # \"Carrier\":\"WN\"\n",
    "    StructField(\"DayOfMonth\", IntegerType(), True), # \"DayOfMonth\":31\n",
    "    StructField(\"DayOfWeek\", IntegerType(), True),  # \"DayOfWeek\":4\n",
    "    StructField(\"DayOfYear\", IntegerType(), True),  # \"DayOfYear\":365\n",
    "    StructField(\"DepDelay\", DoubleType(), True),     # \"DepDelay\":14.0\n",
    "    StructField(\"Dest\", StringType(), True),        # \"Dest\":\"SAN\"\n",
    "    StructField(\"Distance\", DoubleType(), True),     # \"Distance\":368.0\n",
    "    StructField(\"FlightDate\", DateType(), True),    # \"FlightDate\":\"2015-12-30T16:00:00.000-08:00\"\n",
    "    StructField(\"FlightNum\", StringType(), True),   # \"FlightNum\":\"6109\"\n",
    "    StructField(\"Origin\", StringType(), True),      # \"Origin\":\"TUS\"\n",
    "])\n",
    "\n",
    "input_path = \"../data/simple_flight_delay_features.jsonl\"\n",
    "features = spark.read.json(input_path, schema=schema)\n",
    "\n",
    "# Sample 10% to make executable inside the notebook\n",
    "features = features.sample(False, 0.1)\n",
    "\n",
    "features.show(3)\n",
    "features.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Check for nulls in features before using Spark ML\n",
    "#\n",
    "# null_counts = [(column, features.where(features[column].isNull()).count()) for column in features.columns]\n",
    "# cols_with_nulls = filter(lambda x: x[1] > 0, null_counts)\n",
    "# print(\"Columns with nulls that need to be filtered: {}\".format(\n",
    "#     str(list(cols_with_nulls))\n",
    "# ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|HOU-CRP|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|HOU-DAL|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|HOU-DAL|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Add a Route variable to replace FlightNum\n",
    "#\n",
    "from pyspark.sql.functions import lit, concat\n",
    "features_with_route = features.withColumn(\n",
    "  'Route',\n",
    "  concat(\n",
    "    features.Origin,\n",
    "    lit('-'),\n",
    "    features.Dest\n",
    "  )\n",
    ")\n",
    "features_with_route.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------+\n",
      "|ArrDelay|ArrDelayBucket|\n",
      "+--------+--------------+\n",
      "|    43.0|           3.0|\n",
      "|     6.0|           2.0|\n",
      "|    18.0|           2.0|\n",
      "|   -13.0|           1.0|\n",
      "|    -7.0|           1.0|\n",
      "+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay into on-time, slightly late, very late (0, 1, 2)\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "# Setup the Bucketizer\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "arrival_bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "\n",
    "# Save the model\n",
    "arrival_bucketizer_path = \"../models/arrival_bucketizer_2.0.bin\"\n",
    "arrival_bucketizer.write().overwrite().save(arrival_bucketizer_path)\n",
    "\n",
    "# Apply the model\n",
    "ml_bucketized_features = arrival_bucketizer.transform(features_with_route)\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Indexing column \"Carrier\" ...\n",
      "Indexing column \"DayOfMonth\" ...\n",
      "Indexing column \"DayOfWeek\" ...\n",
      "Indexing column \"DayOfYear\" ...\n",
      "Indexing column \"Origin\" ...\n",
      "Indexing column \"Dest\" ...\n",
      "Indexing column \"Route\" ...\n",
      "Indexed all string columns!\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into indexes\n",
    "for column in [\"Carrier\", \"DayOfMonth\", \"DayOfWeek\", \"DayOfYear\",\n",
    "             \"Origin\", \"Dest\", \"Route\"]:\n",
    "    \n",
    "    print(\"Indexing column \\\"{}\\\" ...\".format(column))\n",
    "    \n",
    "    string_indexer = StringIndexer(\n",
    "      inputCol=column,\n",
    "      outputCol=column + \"_index\"\n",
    "    )\n",
    "\n",
    "    string_indexer_model = string_indexer.fit(ml_bucketized_features)\n",
    "    ml_bucketized_features = string_indexer_model.transform(ml_bucketized_features)\n",
    "\n",
    "    # Drop the original column\n",
    "    ml_bucketized_features = ml_bucketized_features.drop(column)\n",
    "\n",
    "    # Save the pipeline model\n",
    "    string_indexer_output_path = \"../models/string_indexer_model_{}.bin\".format(\n",
    "      column\n",
    "    )\n",
    "    string_indexer_model.write().overwrite().save(string_indexer_output_path)\n",
    "\n",
    "print(\"Indexed all string columns!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+--------+--------+----------+---------+--------------+--------------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|DepDelay|Distance|FlightDate|FlightNum|ArrDelayBucket|        Features_vec|\n",
      "+--------+-------------------+-------------------+--------+--------+----------+---------+--------------+--------------------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|    35.0|   187.0|2015-01-02|     1981|           3.0|[35.0,187.0,0.0,1...|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     7.0|   239.0|2015-01-02|       24|           2.0|[7.0,239.0,0.0,1....|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|    27.0|   239.0|2015-01-02|       52|           2.0|[27.0,239.0,0.0,1...|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|    12.0|   883.0|2015-01-02|     1705|           1.0|[12.0,883.0,0.0,1...|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|    -3.0|   571.0|2015-01-02|     4207|           1.0|[-3.0,571.0,0.0,1...|\n",
      "+--------+-------------------+-------------------+--------+--------+----------+---------+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Handle continuous, numeric fields by combining them into one feature vector\n",
    "numeric_columns = [\"DepDelay\", \"Distance\"]\n",
    "index_columns = [\"Carrier_index\", \"DayOfMonth_index\",\n",
    "                 \"DayOfWeek_index\", \"DayOfYear_index\", \"Origin_index\",\n",
    "                 \"Origin_index\", \"Dest_index\", \"Route_index\"]\n",
    "vector_assembler = VectorAssembler(\n",
    "    inputCols=numeric_columns + index_columns,\n",
    "    outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Save the numeric vector assembler\n",
    "vector_assembler_path = \"../models/numeric_vector_assembler.bin\"\n",
    "vector_assembler.write().overwrite().save(vector_assembler_path)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "    final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Inspect the finalized features\n",
    "final_vectorized_features.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementing A More Rigorous Experiment\n",
    "\n",
    "In order to be confident in our experiment for each measure, we need to repeat it at least twice to see how it varies. This is the degree to which we cross-validate. In addition, we need to loop and run the measurement code once for each score. Once we’ve collected several scores for each metric, we look at both the average and standard deviation for each score. Taken together, these scores give us a picture of the quality of our classifier.\n",
    "\n",
    "To begin, we need to iterate and repeat our experiment N times. For each experiment we need to compute a test/train split, then we need to train the model on the training data and apply it to the test data. Then we use `MulticlassClassificationEvaluator` to get a score, once for each metric. We gather the scores in a list for each metric, which we will evaluate at the end of the experiment:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Run 1 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5759860155140392\n",
      "weightedPrecision = 0.6359357493299552\n",
      "weightedRecall = 0.575986015514039\n",
      "f1 = 0.5119318485038695\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5807802512673573\n",
      "weightedPrecision = 0.661402729667335\n",
      "weightedRecall = 0.5807802512673573\n",
      "f1 = 0.5169955455480277\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.591925331018016\n",
      "weightedPrecision = 0.6823700454436153\n",
      "weightedRecall = 0.5919253310180161\n",
      "f1 = 0.5290098787205513\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(\"Run {} out of {} of test/train splits in cross validation...\".format(\n",
    "      i,\n",
    "      split_count,\n",
    "    )\n",
    "  )\n",
    "\n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "\n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4657,\n",
    "    maxMemoryInMB=1024\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "\n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "\n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "  \n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    \n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "\n",
    "    scores[metric_name].append(score)\n",
    "    print(\"{} = {}\".format(metric_name, score))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Processing Run Results\n",
    "\n",
    "Our run leaves us with a `defaultdict` of scores, with one list for each metric. Now we need to compute the average and standard deviation of each list to give us the overall average and standard deviation of each metric:\n",
    "\n",
    "Note that we need to compute both the average and standard deviation of each metric from our run. The average will tell us the approximate performance level, and the standard deviation will tell us how much a model's performance varies. Less variance is desirable. We'll use this information in tuning our model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Log\n",
      "--------------\n",
      "Metric               Average         STD\n",
      "-----------------  ---------  ----------\n",
      "accuracy            0.582897  0.00667715\n",
      "weightedPrecision   0.659903  0.0189864\n",
      "weightedRecall      0.582897  0.00667715\n",
      "f1                  0.519312  0.00716197\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Evaluate average and STD of each metric and print a table\n",
    "#\n",
    "import numpy as np\n",
    "score_averages = defaultdict(float)\n",
    "\n",
    "# Compute the table data\n",
    "average_stds = [] # ha\n",
    "for metric_name in metric_names:\n",
    "  metric_scores = scores[metric_name]\n",
    "  \n",
    "  average_accuracy = sum(metric_scores) / len(metric_scores)\n",
    "  score_averages[metric_name] = average_accuracy\n",
    "  \n",
    "  std_accuracy = np.std(metric_scores)\n",
    "  \n",
    "  average_stds.append((metric_name, average_accuracy, std_accuracy))\n",
    "\n",
    "# Print the table\n",
    "print(\"\\nExperiment Log\")\n",
    "print(\"--------------\")\n",
    "print(tabulate(average_stds, headers=[\"Metric\", \"Average\", \"STD\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "The standard deviations indicate that we might not even need to perform k-fold cross-validation, but an inspection of the underlying scores says otherwise:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(list,\n",
       "            {'accuracy': [0.5759860155140392,\n",
       "              0.5807802512673573,\n",
       "              0.591925331018016],\n",
       "             'weightedPrecision': [0.6359357493299552,\n",
       "              0.661402729667335,\n",
       "              0.6823700454436153],\n",
       "             'weightedRecall': [0.575986015514039,\n",
       "              0.5807802512673573,\n",
       "              0.5919253310180161],\n",
       "             'f1': [0.5119318485038695,\n",
       "              0.5169955455480277,\n",
       "              0.5290098787205513]})"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "scores"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There is actually significant variation between runs, and this could obscure a small improvement (or degradation) in prediction quality.\n",
    "\n",
    "The iterations take time, and this discourages experimentation. A middle ground should be found."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Comparing Experiments to Determine Improvements\n",
    "\n",
    "Now that we have our baseline metrics, we can repeat this code as we improve the model and see what the effect is in terms of the four metrics available to us. So it seems we are done, that we can start playing with our model and features. However, we will quickly run into a problem. We will lose track of the score from the previous run, printed on the screen above many logs for each run, unless we write it down each time. And this is tedious. So, we need to automate this process.\n",
    "\n",
    "What we need to do is load a score log from disk, evaluate the current score in terms of the previous one, and store a new entry to the log back to disk for the next run to access. The following code achieves this aim.\n",
    "\n",
    "First we use pickle to load any existing score log. If this is not present, we initialize a new log, which is simply an empty Python list. Next we prepare the new log entry—a simple Python dict containing the average score for each of four metrics. Then we subtract the previous run’s score to determine the change in this run. This is the information we use to evaluate whether our change worked or not (along with any changes in feature importances, which we will address as well). \n",
    "\n",
    "Finally, we append the new score entry to the log and store it back to disk:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Report\n",
      "-----------------\n",
      "Metric                   Score\n",
      "-----------------  -----------\n",
      "accuracy           -0.00605381\n",
      "weightedPrecision   0.138394\n",
      "weightedRecall     -0.00605381\n",
      "f1                 -0.00615082\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Persist the score to a sccore log that exists between runs\n",
    "#\n",
    "import pickle\n",
    "\n",
    "# Load the score log or initialize an empty one\n",
    "try:\n",
    "  score_log_filename = \"../models/score_log.pickle\"\n",
    "  score_log = pickle.load(open(score_log_filename, \"rb\"))\n",
    "  if not isinstance(score_log, list):\n",
    "    score_log = []\n",
    "except IOError:\n",
    "  score_log = []\n",
    "\n",
    "# Compute the existing score log entry\n",
    "score_log_entry = {metric_name: score_averages[metric_name] for metric_name in metric_names}\n",
    "\n",
    "# Compute and display the change in score for each metric\n",
    "try:\n",
    "  last_log = score_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_log = score_log_entry\n",
    "\n",
    "experiment_report = []\n",
    "for metric_name in metric_names:\n",
    "  run_delta = score_log_entry[metric_name] - last_log[metric_name]\n",
    "  experiment_report.append((metric_name, run_delta))\n",
    "\n",
    "print(\"\\nExperiment Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(experiment_report, headers=[\"Metric\", \"Score\"]))\n",
    "\n",
    "# Append the existing average scores to the log\n",
    "score_log.append(score_log_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(score_log, open(score_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now when we run our script, we will get a report that shows the change between this run and the last run. We can use this, along with our feature importances, to direct our efforts at improving the model. For instance, an example test run shows the model accuracy increase by .003:\n",
    "\n",
    "```\n",
    "Experiment Report\n",
    "-----------------\n",
    "Metric                   Score\n",
    "-----------------  -----------\n",
    "accuracy            0.00300548\n",
    "weightedPrecision  -0.00592227\n",
    "weightedRecall      0.00300548\n",
    "f1                 -0.0105553\n",
    "```\n",
    "\n",
    "Jump back to the code for the model, the code under the section `Implementing a More Rigorous Experiment`. Re-run all the code between there and here, the last three code blocks. See how the score changed slightly? You will use these changes to guide you as you change the model!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inspecting Changes in Feature Importance\n",
    "\n",
    "We can use the list of columns given to our final `VectorAssembler` along with `RandomForestClassificationModel.featureImportances` to derive the importance of each named feature. This is extremely valuable, because like with our prediction quality scores, we can look at changes in feature importances for all features between runs. If a newly introduced feature turns out to be important, it is usually worth adding to the model, so long as it doesn’t hurt quality.\n",
    "\n",
    "### Logging Feature Importances\n",
    "\n",
    "We begin by altering our experiment loop to record feature importances for each run. Check out the abbreviated content from [ch09/improved_spark_mllib_model.py](improved_spark_mllib_model.py):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Run 1 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5848054622304107\n",
      "weightedPrecision = 0.6558041127536607\n",
      "weightedRecall = 0.5848054622304107\n",
      "f1 = 0.5218379487837884\n",
      "\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.587204108977222\n",
      "weightedPrecision = 0.6188744426630224\n",
      "weightedRecall = 0.587204108977222\n",
      "f1 = 0.5227055173394767\n",
      "\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5851237866724833\n",
      "weightedPrecision = 0.6581273095536212\n",
      "weightedRecall = 0.5851237866724833\n",
      "f1 = 0.5186316300276008\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "feature_importances = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(f\"\\nRun {i} out of {split_count} of test/train splits in cross validation...\")\n",
    "\n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "\n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4657,\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "\n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "\n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "\n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "  \n",
    "    scores[metric_name].append(score)\n",
    "    print(f\"{metric_name} = {score}\")\n",
    "\n",
    "  #\n",
    "  # Collect feature importances\n",
    "  #\n",
    "  feature_names = vector_assembler.getInputCols()\n",
    "  feature_importance_list = model.featureImportances\n",
    "  for feature_name, feature_importance in zip(feature_names, feature_importance_list):\n",
    "    feature_importances[feature_name].append(feature_importance)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inspecting Feature Importances\n",
    "\n",
    "Next, we need to compute the average of the importance for each feature. Note that we use a `defaultdict(float)` to ensure that accessing empty keys returns zero. This will be important when comparing entries in the log with different sets of features. In order to print the feature importances, we need to sort them first, by descending order of importance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importances\n",
      "-------------------\n",
      "Name                Importance\n",
      "----------------  ------------\n",
      "DepDelay            0.870979\n",
      "DayOfYear_index     0.0302338\n",
      "Route_index         0.0193866\n",
      "DayOfMonth_index    0.0186692\n",
      "Origin_index        0.0185116\n",
      "Distance            0.0080768\n",
      "Dest_index          0.00800071\n",
      "Carrier_index       0.00515996\n",
      "DayOfWeek_index     0.00247118\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Analyze and report feature importance changes\n",
    "#\n",
    "\n",
    "# Compute averages for each feature\n",
    "feature_importance_entry = defaultdict(float)\n",
    "for feature_name, value_list in feature_importances.items():\n",
    "  average_importance = sum(value_list) / len(value_list)\n",
    "  feature_importance_entry[feature_name] = average_importance\n",
    "\n",
    "# Sort the feature importances in descending order and print\n",
    "import operator\n",
    "sorted_feature_importances = sorted(\n",
    "  feature_importance_entry.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "print(\"\\nFeature Importances\")\n",
    "print(\"-------------------\")\n",
    "print(tabulate(sorted_feature_importances, headers=['Name', 'Importance']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feature Importance Differences Between Runs\n",
    "\n",
    "Next we need to perform the same housekeeping as we did for the model score log: load the model, create an entry for this experiment, load the last experiment and compute the change for each feature between that experiment and the current one, and then print a report on these deltas.\n",
    "\n",
    "First we load the last feature log. If it isn’t available because it doesn’t exist, we initialize the last_feature_log with zeros for each feature, so that new features will have a positive score equal to their amount:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# Compare this run's feature importances with the previous run's\n",
    "#\n",
    "  \n",
    "# Load the feature importance log or initialize an empty one\n",
    "try:\n",
    "  feature_log_filename = \"../models/feature_log.pickle\"\n",
    "  feature_log = pickle.load(open(feature_log_filename, \"rb\"))\n",
    "  if not isinstance(feature_log, list):\n",
    "    feature_log = []\n",
    "except IOError:\n",
    "  feature_log = []\n",
    "\n",
    "# Compute and display the change in score for each feature\n",
    "try:\n",
    "  last_feature_log = feature_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_feature_log = defaultdict(float)\n",
    "  for feature_name, importance in feature_importance_entry.items():\n",
    "    last_feature_log[feature_name] = importance"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we compute the change between the last run and the current one:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Compute the deltas\n",
    "feature_deltas = {}\n",
    "for feature_name in feature_importances.keys():\n",
    "  run_delta = feature_importance_entry[feature_name] - last_feature_log[feature_name]\n",
    "  feature_deltas[feature_name] = run_delta"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to display them, we need to sort the feature importance changes in descending order, to show the biggest change first:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sort feature deltas, biggest change first\n",
    "import operator\n",
    "sorted_feature_deltas = sorted(\n",
    "  feature_deltas.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then we display the sorted feature deltas:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importance Delta Report\n",
      "-------------------------------\n",
      "Feature                  Delta\n",
      "----------------  ------------\n",
      "DepDelay           0.037808\n",
      "DayOfYear_index    0.0302338\n",
      "DayOfMonth_index   0.0186692\n",
      "Origin_index       0.00611677\n",
      "DayOfWeek_index    0.00247118\n",
      "Carrier_index      0.000253099\n",
      "Dest_index        -0.00162545\n",
      "Distance          -0.00389137\n",
      "Route_index       -0.00451193\n"
     ]
    }
   ],
   "source": [
    "# Display sorted feature deltas\n",
    "print(\"\\nFeature Importance Delta Report\")\n",
    "print(\"-------------------------------\")\n",
    "print(tabulate(sorted_feature_deltas, headers=[\"Feature\", \"Delta\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, as with the score log, we append our entry to the log and save it for the next run:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Append the existing average deltas to the log\n",
    "feature_log.append(feature_importance_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(feature_log, open(feature_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We’ll use the raw feature importances as well as the changes in feature importance to guide our creation or alteration of features as we improve the model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Conclusion\n",
    "\n",
    "Now that we have the ability to understand the effect of changes between experimental runs, we can detect changes that improve our model. We can start adding features to test their effect on the model’s prediction quality, and pursue related features that help improve quality! Without this setup, we would be hard put to make positive changes. With it, we are only bounded by our creativity in our efforts to improve the model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Time of Day as a Feature\n",
    "\n",
    "In examining our feature importances, it looks like the date/time fields have some impact. What if we extracted the hour/minute as an integer from the datetime for departure/arrival fields? This would inform the model about morning versus afternoon versus red-eye flights, which surely affects on-time performance, as there is more traffic in the morning than overnight.\n",
    "\n",
    "Check out [ch09/explore_delays.py](explore_delays.py). Let’s start by exploring the premise of this feature, that lateness varies by the time of day of the flight:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|     WN|         2|        5|        2|    12.0| DEN|   883.0|2015-01-02|     1705|   HOU|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|     WN|         2|        5|        2|    -3.0| ECP|   571.0|2015-01-02|     4207|   HOU|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "features.registerTempTable(\"features\")\n",
    "features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------------+---------------------+\n",
      "|Hour|      avg(ArrDelay)|stddev_samp(ArrDelay)|\n",
      "+----+-------------------+---------------------+\n",
      "|   1| -3.893617021276596|   20.182034674166754|\n",
      "|   2|  5.756756756756757|   30.962706425459473|\n",
      "|   3| 13.416666666666666|    64.53393516498733|\n",
      "|   4|               23.0|   23.065125189341593|\n",
      "|   5|                0.4|   10.737783756436894|\n",
      "|   6|-2.7504187604690116|    30.35433397831846|\n",
      "|   7|-0.4401114206128134|   40.152195601591316|\n",
      "|   8|-0.9037614368010843|    35.92628360467317|\n",
      "|   9| 0.8433587786259542|    33.08585785402741|\n",
      "|  10| 2.0286701208981004|    29.64847389458358|\n",
      "|  11|   5.95457650273224|    45.82743048946165|\n",
      "|  12|  4.779426993441491|    36.68864467963427|\n",
      "|  13|  5.700393278512692|     42.5132939301283|\n",
      "|  14|  6.029696578437702|    35.20577118602809|\n",
      "|  15|  7.701895043731779|    38.17410348000706|\n",
      "|  16|  7.808480565371025|    37.93224875212213|\n",
      "|  17|  8.223712835387962|   39.280152086102596|\n",
      "|  18|    8.9209726443769|    41.85564041446721|\n",
      "|  19| 10.703585657370517|    40.95852079310993|\n",
      "|  20| 11.647607934655776|   46.025372518666515|\n",
      "|  21|  9.797132235793946|   42.199556940941086|\n",
      "|  22| 10.248484848484848|   42.130564387689425|\n",
      "|  23|  5.594294770206022|   30.682815092521444|\n",
      "|  24|  4.632183908045977|   36.934489284051836|\n",
      "+----+-------------------+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "  SELECT\n",
    "    HOUR(CRSDepTime) + 1 AS Hour,\n",
    "    AVG(ArrDelay),\n",
    "    STD(ArrDelay)\n",
    "  FROM features\n",
    "  GROUP BY HOUR(CRSDepTime)\n",
    "  ORDER BY HOUR(CRSDepTime)\n",
    "\"\"\").show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+---------------------+\n",
      "|Hour|       avg(ArrDelay)|stddev_samp(ArrDelay)|\n",
      "+----+--------------------+---------------------+\n",
      "|   1|            12.96875|    44.08554264759498|\n",
      "|   2|   8.268292682926829|    39.98876366571234|\n",
      "|   3|                11.5|    41.98743198201629|\n",
      "|   4|               -1.75|   22.704124575223613|\n",
      "|   5|   5.344827586206897|    37.53199885016851|\n",
      "|   6|  1.8566037735849057|     34.6423584069808|\n",
      "|   7|-0.01263157894736842|   31.149900414010926|\n",
      "|   8|  1.0947054436987322|    48.90292678425451|\n",
      "|   9| -0.2966728280961183|    34.74757754418654|\n",
      "|  10| -1.0765265662172878|   29.805160166316067|\n",
      "|  11|  1.0594289508632138|    35.27153464904483|\n",
      "|  12|  1.4428259286234524|    39.81548299617712|\n",
      "|  13|   4.062521008403361|    34.01633558228581|\n",
      "|  14|   4.841894353369764|   39.586194945073466|\n",
      "|  15|   5.102511880515954|    37.92038412105885|\n",
      "|  16|   5.191109555477774|   35.272496914442854|\n",
      "|  17|   6.571292775665399|    35.76201031340733|\n",
      "|  18|   8.639368895567243|    43.13049596430492|\n",
      "|  19|   8.666666666666666|    39.81014321100816|\n",
      "|  20|   9.608771929824561|    38.50663894431138|\n",
      "|  21|  10.441700960219478|    46.51739411355749|\n",
      "|  22|   9.639167595689335|   45.578718342016785|\n",
      "|  23|   9.668339416058394|   39.350731791672594|\n",
      "|  24|   10.72784412319296|   40.562252462679226|\n",
      "+----+--------------------+---------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "  SELECT\n",
    "    HOUR(CRSArrTime) + 1 AS Hour,\n",
    "    AVG(ArrDelay),\n",
    "    STD(ArrDelay)\n",
    "  FROM features\n",
    "  GROUP BY HOUR(CRSArrTime)\n",
    "  ORDER BY HOUR(CRSArrTime)\n",
    "\"\"\").show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Departure delay covariance: 16.66980794454114\n",
      "Arrival delay covariance:   15.940080898824531\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import hour\n",
    "\n",
    "features = features.withColumn('CRSDepHourOfDay', hour(features.CRSDepTime))\n",
    "features = features.withColumn('CRSArrHourOfDay', hour(features.CRSArrTime))\n",
    "\n",
    "departure_cov = features.stat.cov('CRSDepHourOfDay', 'ArrDelay')\n",
    "arrival_cov = features.stat.cov('CRSArrHourOfDay', 'ArrDelay')\n",
    "\n",
    "print(\"Departure delay covariance: {:,}\".format(departure_cov))\n",
    "print(\"Arrival delay covariance:   {:,}\".format(arrival_cov))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------------+-------------------+---------------+\n",
      "|         CRSDepTime|CRSDepHourOfDay|         CRSArrTime|CRSArrHourOfDay|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "|2015-01-02 22:05:00|             22|2015-01-02 22:55:00|             22|\n",
      "|2015-01-02 12:00:00|             12|2015-01-02 12:55:00|             12|\n",
      "|2015-01-02 19:00:00|             19|2015-01-02 20:00:00|             20|\n",
      "|2015-01-02 06:00:00|              6|2015-01-02 07:40:00|              7|\n",
      "|2015-01-02 10:20:00|             10|2015-01-02 11:55:00|             11|\n",
      "|2015-01-02 14:10:00|             14|2015-01-02 15:10:00|             15|\n",
      "|2015-01-02 10:20:00|             10|2015-01-02 13:40:00|             13|\n",
      "|2015-01-02 17:00:00|             17|2015-01-02 18:45:00|             18|\n",
      "|2015-01-02 07:35:00|              7|2015-01-02 10:45:00|             10|\n",
      "|2015-01-02 05:50:00|              5|2015-01-02 08:25:00|              8|\n",
      "|2015-01-02 09:25:00|              9|2015-01-02 10:25:00|             10|\n",
      "|2015-01-02 08:50:00|              8|2015-01-02 11:10:00|             11|\n",
      "|2015-01-02 20:20:00|             20|2015-01-02 23:45:00|             23|\n",
      "|2015-01-02 19:30:00|             19|2015-01-02 21:20:00|             21|\n",
      "|2015-01-02 06:30:00|              6|2015-01-02 07:30:00|              7|\n",
      "|2015-01-02 12:00:00|             12|2015-01-02 12:45:00|             12|\n",
      "|2015-01-02 20:20:00|             20|2015-01-02 21:15:00|             21|\n",
      "|2015-01-02 13:55:00|             13|2015-01-02 17:00:00|             17|\n",
      "|2015-01-02 08:00:00|              8|2015-01-02 10:45:00|             10|\n",
      "|2015-01-02 11:45:00|             11|2015-01-02 16:45:00|             16|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "features.select(\n",
    "    \"CRSDepTime\", \n",
    "    \"CRSDepHourOfDay\", \n",
    "    \"CRSArrTime\", \n",
    "    \"CRSArrHourOfDay\"\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Encoding Our New Features\n",
    "\n",
    "Now we must repeat the feature encoding process such that it includes these new features. Lets take a look at what our features look like at this moment:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|CRSDepHourOfDay|CRSArrHourOfDay|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|             22|             22|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|             12|             12|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|             19|             20|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|     WN|         2|        5|        2|    12.0| DEN|   883.0|2015-01-02|     1705|   HOU|              6|              7|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|     WN|         2|        5|        2|    -3.0| ECP|   571.0|2015-01-02|     4207|   HOU|             10|             11|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "features.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So we're back at the beginning, and still have to add `Route`, bucketize the data, encode our string and numeric fields and then combine them into a single vector. Lets get started!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|CRSDepHourOfDay|CRSArrHourOfDay|  Route|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|             22|             22|HOU-CRP|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|             12|             12|HOU-DAL|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|             19|             20|HOU-DAL|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|     WN|         2|        5|        2|    12.0| DEN|   883.0|2015-01-02|     1705|   HOU|              6|              7|HOU-DEN|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|     WN|         2|        5|        2|    -3.0| ECP|   571.0|2015-01-02|     4207|   HOU|             10|             11|HOU-ECP|\n",
      "|   -14.0|2015-01-02 15:10:00|2015-01-02 14:10:00|     WN|         2|        5|        2|     0.0| ELP|   677.0|2015-01-02|     1258|   HOU|             14|             15|HOU-ELP|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+\n",
      "only showing top 6 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Add a Route variable to replace FlightNum\n",
    "#\n",
    "from pyspark.sql.functions import lit, concat\n",
    "features_with_route = features.withColumn(\n",
    "  'Route',\n",
    "  concat(\n",
    "    features.Origin,\n",
    "    lit('-'),\n",
    "    features.Dest\n",
    "  )\n",
    ")\n",
    "features_with_route.show(6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------+\n",
      "|ArrDelay|ArrDelayBucket|\n",
      "+--------+--------------+\n",
      "|    43.0|           3.0|\n",
      "|     6.0|           2.0|\n",
      "|    18.0|           2.0|\n",
      "|   -13.0|           1.0|\n",
      "|    -7.0|           1.0|\n",
      "+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay into on-time, slightly late, very late (0, 1, 2)\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "# Setup the Bucketizer\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "arrival_bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "\n",
    "# Save the model\n",
    "arrival_bucketizer_path = \"../models/arrival_bucketizer_2.0.bin\"\n",
    "arrival_bucketizer.write().overwrite().save(arrival_bucketizer_path)\n",
    "\n",
    "# Apply the model\n",
    "ml_bucketized_features = arrival_bucketizer.transform(features_with_route)\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+-------------+----------------+---------------+---------------+------------+----------+-----------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|CRSDepHourOfDay|CRSArrHourOfDay|  Route|ArrDelayBucket|Carrier_index|DayOfMonth_index|DayOfWeek_index|DayOfYear_index|Origin_index|Dest_index|Route_index|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+-------------+----------------+---------------+---------------+------------+----------+-----------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|             22|             22|HOU-CRP|           3.0|          0.0|             1.0|            0.0|            1.0|        28.0|     102.0|      938.0|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|             12|             12|HOU-DAL|           2.0|          0.0|             1.0|            0.0|            1.0|        28.0|      27.0|       55.0|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|             19|             20|HOU-DAL|           2.0|          0.0|             1.0|            0.0|            1.0|        28.0|      27.0|       55.0|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|     WN|         2|        5|        2|    12.0| DEN|   883.0|2015-01-02|     1705|   HOU|              6|              7|HOU-DEN|           1.0|          0.0|             1.0|            0.0|            1.0|        28.0|       3.0|     1089.0|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|     WN|         2|        5|        2|    -3.0| ECP|   571.0|2015-01-02|     4207|   HOU|             10|             11|HOU-ECP|           1.0|          0.0|             1.0|            0.0|            1.0|        28.0|     151.0|     2673.0|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+-------------+----------------+---------------+---------------+------------+----------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into indexes\n",
    "for column in [\"Carrier\", \"DayOfMonth\", \"DayOfWeek\", \"DayOfYear\",\n",
    "               \"Origin\", \"Dest\", \"Route\"]:\n",
    "  string_indexer = StringIndexer(\n",
    "    inputCol=column,\n",
    "    outputCol=column + \"_index\"\n",
    "  )\n",
    "\n",
    "  string_indexer_model = string_indexer.fit(ml_bucketized_features)\n",
    "  ml_bucketized_features = string_indexer_model.transform(ml_bucketized_features)\n",
    "\n",
    "  # Save the pipeline model\n",
    "  string_indexer_output_path = \"../models/string_indexer_model_3.0.{}.bin\".format(\n",
    "    column\n",
    "  )\n",
    "  string_indexer_model.write().overwrite().save(string_indexer_output_path)\n",
    "\n",
    "ml_bucketized_features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+--------------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|CRSDepHourOfDay|CRSArrHourOfDay|  Route|ArrDelayBucket|        Features_vec|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+--------------------+\n",
      "|    43.0|2015-01-02 22:55:00|2015-01-02 22:05:00|     WN|         2|        5|        2|    35.0| CRP|   187.0|2015-01-02|     1981|   HOU|             22|             22|HOU-CRP|           3.0|[35.0,187.0,22.0,...|\n",
      "|     6.0|2015-01-02 12:55:00|2015-01-02 12:00:00|     WN|         2|        5|        2|     7.0| DAL|   239.0|2015-01-02|       24|   HOU|             12|             12|HOU-DAL|           2.0|[7.0,239.0,12.0,1...|\n",
      "|    18.0|2015-01-02 20:00:00|2015-01-02 19:00:00|     WN|         2|        5|        2|    27.0| DAL|   239.0|2015-01-02|       52|   HOU|             19|             20|HOU-DAL|           2.0|[27.0,239.0,19.0,...|\n",
      "|   -13.0|2015-01-02 07:40:00|2015-01-02 06:00:00|     WN|         2|        5|        2|    12.0| DEN|   883.0|2015-01-02|     1705|   HOU|              6|              7|HOU-DEN|           1.0|[12.0,883.0,6.0,7...|\n",
      "|    -7.0|2015-01-02 11:55:00|2015-01-02 10:20:00|     WN|         2|        5|        2|    -3.0| ECP|   571.0|2015-01-02|     4207|   HOU|             10|             11|HOU-ECP|           1.0|[-3.0,571.0,10.0,...|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+---------------+---------------+-------+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Combine continuous, numeric fields with indexes of nominal ones\n",
    "# ...into one feature vector\n",
    "numeric_columns = [\n",
    "  \"DepDelay\", \"Distance\",\n",
    "  \"CRSDepHourOfDay\", \"CRSArrHourOfDay\"\n",
    "]\n",
    "index_columns = [\"Carrier_index\", \"DayOfMonth_index\",\n",
    "                 \"DayOfWeek_index\", \"DayOfYear_index\", \"Origin_index\",\n",
    "                 \"Origin_index\", \"Dest_index\", \"Route_index\"]\n",
    "vector_assembler = VectorAssembler(\n",
    "  inputCols=numeric_columns + index_columns,\n",
    "  outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Save the numeric vector assembler\n",
    "vector_assembler_path = \"../models/numeric_vector_assembler_3.0.bin\"\n",
    "vector_assembler.write().overwrite().save(vector_assembler_path)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "  final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Inspect the finalized features\n",
    "final_vectorized_features.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Training with Our New Features\n",
    "\n",
    "Now we will train the model again, noting the performance and feature importances as we did before. This allows us to see the change in performance owing to the introduction of these new fields, `CRSDepHourOfDay` and `CRSArrHourOfDay`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Run 1 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5866520787746171\n",
      "weightedPrecision = 0.6990370807802881\n",
      "weightedRecall = 0.586652078774617\n",
      "f1 = 0.5217345660799317\n",
      "\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5902031063321386\n",
      "weightedPrecision = 0.6507855345185848\n",
      "weightedRecall = 0.5902031063321386\n",
      "f1 = 0.5258478731116644\n",
      "\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5827243974261097\n",
      "weightedPrecision = 0.6341957653456587\n",
      "weightedRecall = 0.5827243974261097\n",
      "f1 = 0.5207586903655927\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "feature_importances = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(f\"\\nRun {i} out of {split_count} of test/train splits in cross validation...\")\n",
    "\n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "\n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4657,\n",
    "    maxMemoryInMB=1024\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "\n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "\n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "\n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "  \n",
    "    scores[metric_name].append(score)\n",
    "    print(f\"{metric_name} = {score}\")\n",
    "\n",
    "  #\n",
    "  # Collect feature importances\n",
    "  #\n",
    "  feature_names = vector_assembler.getInputCols()\n",
    "  feature_importance_list = model.featureImportances\n",
    "  for feature_name, feature_importance in zip(feature_names, feature_importance_list):\n",
    "    feature_importances[feature_name].append(feature_importance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Log\n",
      "--------------\n",
      "Metric               Average         STD\n",
      "-----------------  ---------  ----------\n",
      "accuracy            0.586527  0.00305446\n",
      "weightedPrecision   0.661339  0.0275032\n",
      "weightedRecall      0.586527  0.00305446\n",
      "f1                  0.52278   0.00220533\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Evaluate average and STD of each metric and print a table\n",
    "#\n",
    "import numpy as np\n",
    "score_averages = defaultdict(float)\n",
    "\n",
    "# Compute the table data\n",
    "average_stds = [] # ha\n",
    "for metric_name in metric_names:\n",
    "  metric_scores = scores[metric_name]\n",
    "  \n",
    "  average_accuracy = sum(metric_scores) / len(metric_scores)\n",
    "  score_averages[metric_name] = average_accuracy\n",
    "  \n",
    "  std_accuracy = np.std(metric_scores)\n",
    "  \n",
    "  average_stds.append((metric_name, average_accuracy, std_accuracy))\n",
    "\n",
    "# Print the table\n",
    "print(\"\\nExperiment Log\")\n",
    "print(\"--------------\")\n",
    "print(tabulate(average_stds, headers=[\"Metric\", \"Average\", \"STD\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Report\n",
      "-----------------\n",
      "Metric                  Score\n",
      "-----------------  ----------\n",
      "accuracy           0.00362933\n",
      "weightedPrecision  0.00143662\n",
      "weightedRecall     0.00362933\n",
      "f1                 0.00346795\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Persist the score to a sccore log that exists between runs\n",
    "#\n",
    "import pickle\n",
    "\n",
    "# Load the score log or initialize an empty one\n",
    "try:\n",
    "  score_log_filename = \"../models/score_log.pickle\"\n",
    "  score_log = pickle.load(open(score_log_filename, \"rb\"))\n",
    "  if not isinstance(score_log, list):\n",
    "    score_log = []\n",
    "except IOError:\n",
    "  score_log = []\n",
    "\n",
    "# Compute the existing score log entry\n",
    "score_log_entry = {metric_name: score_averages[metric_name] for metric_name in metric_names}\n",
    "\n",
    "# Compute and display the change in score for each metric\n",
    "try:\n",
    "  last_log = score_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_log = score_log_entry\n",
    "\n",
    "experiment_report = []\n",
    "for metric_name in metric_names:\n",
    "  run_delta = score_log_entry[metric_name] - last_log[metric_name]\n",
    "  experiment_report.append((metric_name, run_delta))\n",
    "\n",
    "print(\"\\nExperiment Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(experiment_report, headers=[\"Metric\", \"Score\"]))\n",
    "\n",
    "# Append the existing average scores to the log\n",
    "score_log.append(score_log_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(score_log, open(score_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importances\n",
      "-------------------\n",
      "Name                Importance\n",
      "----------------  ------------\n",
      "DepDelay            0.863795\n",
      "DayOfMonth_index    0.0270982\n",
      "DayOfYear_index     0.0246649\n",
      "Route_index         0.0177811\n",
      "Origin_index        0.0173232\n",
      "Distance            0.0122518\n",
      "Dest_index          0.00733773\n",
      "Carrier_index       0.00651014\n",
      "CRSDepHourOfDay     0.00259898\n",
      "CRSArrHourOfDay     0.00176026\n",
      "DayOfWeek_index     0.00155594\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Analyze and report feature importance changes\n",
    "#\n",
    "\n",
    "# Compute averages for each feature\n",
    "feature_importance_entry = defaultdict(float)\n",
    "for feature_name, value_list in feature_importances.items():\n",
    "  average_importance = sum(value_list) / len(value_list)\n",
    "  feature_importance_entry[feature_name] = average_importance\n",
    "\n",
    "# Sort the feature importances in descending order and print\n",
    "import operator\n",
    "sorted_feature_importances = sorted(\n",
    "  feature_importance_entry.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "print(\"\\nFeature Importances\")\n",
    "print(\"-------------------\")\n",
    "print(tabulate(sorted_feature_importances, headers=['Name', 'Importance']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importance Delta Report\n",
      "-------------------------------\n",
      "Feature                  Delta\n",
      "----------------  ------------\n",
      "DayOfMonth_index   0.00842893\n",
      "Distance           0.00417504\n",
      "CRSDepHourOfDay    0.00259898\n",
      "CRSArrHourOfDay    0.00176026\n",
      "Carrier_index      0.00135018\n",
      "Dest_index        -0.000662988\n",
      "DayOfWeek_index   -0.000915246\n",
      "Origin_index      -0.00118843\n",
      "Route_index       -0.0016055\n",
      "DayOfYear_index   -0.00556882\n",
      "DepDelay          -0.00718399\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Compare this run's feature importances with the previous run's\n",
    "#\n",
    "  \n",
    "# Load the feature importance log or initialize an empty one\n",
    "try:\n",
    "  feature_log_filename = \"../models/feature_log.pickle\"\n",
    "  feature_log = pickle.load(open(feature_log_filename, \"rb\"))\n",
    "  if not isinstance(feature_log, list):\n",
    "    feature_log = []\n",
    "except IOError:\n",
    "  feature_log = []\n",
    "\n",
    "# Compute and display the change in score for each feature\n",
    "try:\n",
    "  last_feature_log = feature_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_feature_log = defaultdict(float)\n",
    "  for feature_name, importance in feature_importance_entry.items():\n",
    "    last_feature_log[feature_name] = importance\n",
    "\n",
    "# Compute the deltas\n",
    "feature_deltas = {}\n",
    "for feature_name in feature_importances.keys():\n",
    "  run_delta = feature_importance_entry[feature_name] - last_feature_log[feature_name]\n",
    "  feature_deltas[feature_name] = run_delta\n",
    "\n",
    "# Sort feature deltas, biggest change first\n",
    "import operator\n",
    "sorted_feature_deltas = sorted(\n",
    "  feature_deltas.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "# Display sorted feature deltas\n",
    "print(\"\\nFeature Importance Delta Report\")\n",
    "print(\"-------------------------------\")\n",
    "print(tabulate(sorted_feature_deltas, headers=[\"Feature\", \"Delta\"]))\n",
    "\n",
    "# Append the existing average deltas to the log\n",
    "feature_log.append(feature_importance_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(feature_log, open(feature_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Interpreting Our Results\n",
    "\n",
    "Interpreting the output, it looks like the combined effect of these fields is to impact feature importance by about 1%, but the effect on accuracy is insignificant. We’ll leave the fields in, although they don’t help much. Without resorting to advanced time series analysis, it seems we’ve milked all we can from date/time-based features.\n",
    "\n",
    "## Incorporating Airplane Data\n",
    "\n",
    "Recall from `Investigating Airplanes (Entities)` that we incorporated data on airplane manufacturers into our data model. For instance, we analyzed the distribution of manufacturers in the American commercial fleet. In this section, we’re going to join in airline data and see what impact this has on the model’s accuracy.\n",
    "\n",
    "I wonder whether properties of the aircraft (called the “metal” of the flight) influence delays? For instance, bigger aircraft fly higher and can go over weather, while smaller aircraft may be less able to do so. I can’t honestly think of a reason why the engine manufacturer, airplane manufacturer, or manufacture year would have an impact on the model, but since we’re importing one field, we may as well try them all! Note that we can simply drop any features that don’t rank as very significant. The beauty of our experimental model with decision trees is that it doesn’t cost extra to try extra fields. Sometimes you can simply let the model decide what matters.\n",
    "\n",
    "Note that when dealing with team members and with other teams who need an accounting of your time in order to coordinate with you, a description of the experiments you are running will help keep the teams in sync. For instance, “We are attempting to incorporate a new dataset which we scraped from the FAA website into our flight delay predictive model” would make a good experimental description during an agile sprint.\n",
    "\n",
    "### Extracting Airplane Features\n",
    "\n",
    "To add airplane features to our model, we need to create a new feature extraction script, [ch09/extract_features_with_airplanes.py](extract_features_with_airplanes.py). We can do this by copying and altering [ch09/extract_features.py](extract_features.py).\n",
    "\n",
    "First we add `TailNum` to the fields we select from our training data. Because this column also appears in our airplane dataset, we need to name it differently or we won’t easily be able to access the column after the join. We’ll name it `FeatureTailNum`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+--------------+\n",
      "|FlightNum|FlightDate|FeatureTailNum|\n",
      "+---------+----------+--------------+\n",
      "|      937|2015-01-01|        N009AA|\n",
      "|     1310|2015-01-01|        N011AA|\n",
      "|     2227|2015-01-01|        N016AA|\n",
      "|      903|2015-01-01|        N017AA|\n",
      "|       17|2015-01-01|        N019AA|\n",
      "|      324|2015-01-01|        N202AA|\n",
      "|     2413|2015-01-01|        N357AA|\n",
      "|       28|2015-01-01|        N358AA|\n",
      "|     1621|2015-01-01|        N367AA|\n",
      "|     1210|2015-01-01|        N397AA|\n",
      "+---------+----------+--------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Load the on-time parquet file\n",
    "input_path = \"../data/january_performance.parquet\"\n",
    "on_time_dataframe = spark.read.parquet(input_path)\n",
    "on_time_dataframe.registerTempTable(\"on_time_performance\")\n",
    "\n",
    "# Select a few features of interest\n",
    "simple_on_time_features = spark.sql(\"\"\"\n",
    "SELECT\n",
    "  FlightNum,\n",
    "  FlightDate,\n",
    "  DayOfWeek,\n",
    "  DayofMonth AS DayOfMonth,\n",
    "  CONCAT(Month, '-',  DayofMonth) AS DayOfYear,\n",
    "  Carrier,\n",
    "  Origin,\n",
    "  Dest,\n",
    "  Distance,\n",
    "  DepDelay,\n",
    "  ArrDelay,\n",
    "  CRSDepTime,\n",
    "  CRSArrTime,\n",
    "  CONCAT(Origin, '-', Dest) AS Route,\n",
    "  TailNum AS FeatureTailNum\n",
    "FROM on_time_performance\n",
    "WHERE FlightDate < '2015-02-01'\n",
    "\"\"\")\n",
    "\n",
    "simple_on_time_features = simple_on_time_features.sample(False, 0.1)\n",
    "\n",
    "simple_on_time_features.select(\n",
    "  \"FlightNum\",\n",
    "  \"FlightDate\",\n",
    "  \"FeatureTailNum\"\n",
    ").show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+--------------+\n",
      "|FlightNum|FlightDate|DayOfWeek|DayOfMonth|DayOfYear|Carrier|Origin|Dest|Distance|DepDelay|ArrDelay|CRSDepTime|CRSArrTime|  Route|FeatureTailNum|\n",
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+--------------+\n",
      "|      937|2015-01-01|        4|         1|      1-1|     AA|   EGE| LAX|   748.0|    10.0|   -12.0|      1805|      1920|EGE-LAX|        N009AA|\n",
      "|     1310|2015-01-01|        4|         1|      1-1|     AA|   DFW| CLE|  1021.0|    47.0|    42.0|      0855|      1225|DFW-CLE|        N011AA|\n",
      "|     2227|2015-01-01|        4|         1|      1-1|     AA|   MIA| IAH|   964.0|    42.0|    49.0|      1640|      1842|MIA-IAH|        N016AA|\n",
      "|      903|2015-01-01|        4|         1|      1-1|     AA|   LAX| EGE|   748.0|    -7.0|    16.0|      0830|      1145|LAX-EGE|        N017AA|\n",
      "|       17|2015-01-01|        4|         1|      1-1|     AA|   ATL| MIA|   594.0|    -4.0|   -10.0|      0700|      0852|ATL-MIA|        N019AA|\n",
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter nulls, they can't help us\n",
    "filled_on_time_features = simple_on_time_features.where(simple_on_time_features.ArrDelay.isNotNull())\n",
    "filled_on_time_features = filled_on_time_features.where(filled_on_time_features.DepDelay.isNotNull())\n",
    "filled_on_time_features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We need to turn timestamps into timestamps, and not strings or numbers\n",
    "def convert_hours(hours_minutes):\n",
    "  hours = hours_minutes[:-2]\n",
    "  minutes = hours_minutes[-2:]\n",
    "  \n",
    "  if hours == '24':\n",
    "    hours = '23'\n",
    "    minutes = '59'\n",
    "  \n",
    "  time_string = \"{}:{}:00Z\".format(hours, minutes)\n",
    "  return time_string\n",
    "\n",
    "def compose_datetime(iso_date, time_string):\n",
    "  return \"{} {}\".format(iso_date, time_string)\n",
    "\n",
    "def create_iso_string(iso_date, hours_minutes):\n",
    "  time_string = convert_hours(hours_minutes)\n",
    "  full_datetime = compose_datetime(iso_date, time_string)\n",
    "  return full_datetime\n",
    "\n",
    "def create_datetime(iso_string):\n",
    "  return iso8601.parse_date(iso_string)\n",
    "\n",
    "def convert_datetime(iso_date, hours_minutes):\n",
    "  iso_string = create_iso_string(iso_date, hours_minutes)\n",
    "  dt = create_datetime(iso_string)\n",
    "  return dt\n",
    "\n",
    "def day_of_year(iso_date_string):\n",
    "  dt = iso8601.parse_date(iso_date_string)\n",
    "  doy = dt.timetuple().tm_yday\n",
    "  return doy\n",
    "\n",
    "def alter_feature_datetimes(row):\n",
    "  \n",
    "  flight_date = iso8601.parse_date(row['FlightDate'])\n",
    "  scheduled_dep_time = convert_datetime(row['FlightDate'], row['CRSDepTime'])\n",
    "  scheduled_arr_time = convert_datetime(row['FlightDate'], row['CRSArrTime'])\n",
    "  \n",
    "  # Handle overnight flights\n",
    "  if scheduled_arr_time < scheduled_dep_time:\n",
    "    scheduled_arr_time += datetime.timedelta(days=1)\n",
    "  \n",
    "  doy = day_of_year(row['FlightDate'])\n",
    "  \n",
    "  return {\n",
    "    'FlightNum': row['FlightNum'],\n",
    "    'FlightDate': flight_date,\n",
    "    'DayOfWeek': int(row['DayOfWeek']),\n",
    "    'DayOfMonth': int(row['DayOfMonth']),\n",
    "    'DayOfYear': doy,\n",
    "    'Carrier': row['Carrier'],\n",
    "    'Origin': row['Origin'],\n",
    "    'Dest': row['Dest'],\n",
    "    'Distance': row['Distance'],\n",
    "    'DepDelay': row['DepDelay'],\n",
    "    'ArrDelay': row['ArrDelay'],\n",
    "    'CRSDepTime': scheduled_dep_time,\n",
    "    'CRSArrTime': scheduled_arr_time,\n",
    "    'Route': row['Route'],\n",
    "    'FeatureTailNum': row['FeatureTailNum'],\n",
    "  }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/vagrant/spark/python/pyspark/sql/session.py:366: UserWarning: Using RDD of dict to inferSchema is deprecated. Use pyspark.sql.Row instead\n",
      "  warnings.warn(\"Using RDD of dict to inferSchema is deprecated. \"\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+--------------+-------------------+---------+------+-------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FeatureTailNum|         FlightDate|FlightNum|Origin|  Route|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+--------------+-------------------+---------+------+-------+\n",
      "|   -12.0|2015-01-01 19:20:00|2015-01-01 18:05:00|     AA|         1|        4|        1|    10.0| LAX|   748.0|        N009AA|2015-01-01 00:00:00|      937|   EGE|EGE-LAX|\n",
      "|    42.0|2015-01-01 12:25:00|2015-01-01 08:55:00|     AA|         1|        4|        1|    47.0| CLE|  1021.0|        N011AA|2015-01-01 00:00:00|     1310|   DFW|DFW-CLE|\n",
      "|    49.0|2015-01-01 18:42:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    42.0| IAH|   964.0|        N016AA|2015-01-01 00:00:00|     2227|   MIA|MIA-IAH|\n",
      "|    16.0|2015-01-01 11:45:00|2015-01-01 08:30:00|     AA|         1|        4|        1|    -7.0| EGE|   748.0|        N017AA|2015-01-01 00:00:00|      903|   LAX|LAX-EGE|\n",
      "|   -10.0|2015-01-01 08:52:00|2015-01-01 07:00:00|     AA|         1|        4|        1|    -4.0| MIA|   594.0|        N019AA|2015-01-01 00:00:00|       17|   ATL|ATL-MIA|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+--------------+-------------------+---------+------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "timestamp_features = filled_on_time_features.rdd.map(alter_feature_datetimes)\n",
    "timestamp_df = timestamp_features.toDF()\n",
    "timestamp_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Joining Airplane Data\n",
    "\n",
    "Next, we load the airplane data and left join it to our features dataset. Note that null is a problematic value for our `StringIndexer`. But we don’t want to discard empty values or rows either, because whether a variable is present or not is something our decision tree model can use to learn. We use `DataFrame.selectExpr` to `COALESCE` our `null` values to the string `'Empty'`. This will get its own index from `StringIndexer` and things will work out well. Also note that we rename `FeatureTailNum` back to `TailNum` for the final output:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-------------------+---------+----------+---------+-------+------+----+--------+--------+--------+-------------------+-------------------+-------+-------+------------------+-----------+------------+----------------+-----+----------+\n",
      "|FlightNum|         FlightDate|DayOfWeek|DayOfMonth|DayOfYear|Carrier|Origin|Dest|Distance|DepDelay|ArrDelay|         CRSDepTime|         CRSArrTime|  Route|TailNum|EngineManufacturer|EngineModel|Manufacturer|ManufacturerYear|Model|OwnerState|\n",
      "+---------+-------------------+---------+----------+---------+-------+------+----+--------+--------+--------+-------------------+-------------------+-------+-------+------------------+-----------+------------+----------------+-----+----------+\n",
      "|      937|2015-01-01 00:00:00|        4|         1|        1|     AA|   EGE| LAX|   748.0|    10.0|   -12.0|2015-01-01 18:05:00|2015-01-01 19:20:00|EGE-LAX| N009AA|             Empty|      Empty|       Empty|           Empty|Empty|     Empty|\n",
      "|     1310|2015-01-01 00:00:00|        4|         1|        1|     AA|   DFW| CLE|  1021.0|    47.0|    42.0|2015-01-01 08:55:00|2015-01-01 12:25:00|DFW-CLE| N011AA|             Empty|      Empty|       Empty|           Empty|Empty|     Empty|\n",
      "|     2227|2015-01-01 00:00:00|        4|         1|        1|     AA|   MIA| IAH|   964.0|    42.0|    49.0|2015-01-01 16:40:00|2015-01-01 18:42:00|MIA-IAH| N016AA|             Empty|      Empty|       Empty|           Empty|Empty|     Empty|\n",
      "|      903|2015-01-01 00:00:00|        4|         1|        1|     AA|   LAX| EGE|   748.0|    -7.0|    16.0|2015-01-01 08:30:00|2015-01-01 11:45:00|LAX-EGE| N017AA|             Empty|      Empty|       Empty|           Empty|Empty|     Empty|\n",
      "|       17|2015-01-01 00:00:00|        4|         1|        1|     AA|   ATL| MIA|   594.0|    -4.0|   -10.0|2015-01-01 07:00:00|2015-01-01 08:52:00|ATL-MIA| N019AA|             Empty|      Empty|       Empty|           Empty|Empty|     Empty|\n",
      "+---------+-------------------+---------+----------+---------+-------+------+----+--------+--------+--------+-------------------+-------------------+-------+-------+------------------+-----------+------------+----------------+-----+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Load airplanes and left join on tail numbers\n",
    "airplanes_path = \"../data/airplanes.jsonl\"\n",
    "airplanes = spark.read.json(airplanes_path)\n",
    "\n",
    "# Left outer join ensures all feature records remain, with nulls where airplane records are not available\n",
    "features_with_airplanes = timestamp_df.join(\n",
    "  airplanes,\n",
    "  on=timestamp_df.FeatureTailNum == airplanes.TailNum,\n",
    "  how=\"left_outer\"\n",
    ")\n",
    "\n",
    "# Fill in the nulls 'Empty' with COALESCE\n",
    "features_with_airplanes = features_with_airplanes.selectExpr(\n",
    "  \"FlightNum\",\n",
    "  \"FlightDate\",\n",
    "  \"DayOfWeek\",\n",
    "  \"DayOfMonth\",\n",
    "  \"DayOfYear\",\n",
    "  \"Carrier\",\n",
    "  \"Origin\",\n",
    "  \"Dest\",\n",
    "  \"Distance\",\n",
    "  \"DepDelay\",\n",
    "  \"ArrDelay\",\n",
    "  \"CRSDepTime\",\n",
    "  \"CRSArrTime\",\n",
    "  \"Route\",\n",
    "  \"FeatureTailNum AS TailNum\",\n",
    "  \"COALESCE(EngineManufacturer, 'Empty') AS EngineManufacturer\",\n",
    "  \"COALESCE(EngineModel, 'Empty') AS EngineModel\",\n",
    "  \"COALESCE(Manufacturer, 'Empty') AS Manufacturer\",\n",
    "  \"COALESCE(ManufacturerYear, 'Empty') AS ManufacturerYear\",\n",
    "  \"COALESCE(Model, 'Empty') AS Model\",\n",
    "  \"COALESCE(OwnerState, 'Empty') AS OwnerState\"\n",
    ")\n",
    "features_with_airplanes.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Explicitly sort the data and keep it sorted throughout. Leave nothing to chance.\n",
    "sorted_features = features_with_airplanes.sort(\n",
    "  timestamp_df.DayOfYear,\n",
    "  timestamp_df.Carrier,\n",
    "  timestamp_df.Origin,\n",
    "  timestamp_df.Dest,\n",
    "  timestamp_df.FlightNum,\n",
    "  timestamp_df.CRSDepTime,\n",
    "  timestamp_df.CRSArrTime,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Storing Our Features\n",
    "\n",
    "Finally, we store the final output to a new path. We’ll have to remember to alter our model training script to point at this new path:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Store as a single json file\n",
    "output_path = \"../data/simple_flight_delay_features_airplanes.json\"\n",
    "sorted_features.repartition(1).write.mode(\"overwrite\").json(output_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we’re ready to incorporate the features into our model.\n",
    "\n",
    "## Incorporating Airplane Features into Our Classifier Model\n",
    "\n",
    "Now we need to create a new script that incorporates our new airplane features into our classifier model. Check out [ch09/spark_model_with_airplanes.py](spark_model_with_airplanes.py), which we copied from [ch09/improved_spark_mllib_model.py](improved_spark_mllib_model.py) and altered.\n",
    "\n",
    "First we need to load the training data with the additional fields, including Route (which is now calculated in [ch09/extract_features_with_airplanes.py](extract_features_with_airplanes.py)):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|EngineManufacturer|EngineModel|     Manufacturer|ManufacturerYear|OwnerState|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+\n",
      "|   -10.0|2015-01-01 08:52:00|2015-01-01 07:00:00|     AA|         1|        4|        1|    -4.0| MIA|   594.0|2015-01-01|       17|   ATL|ATL-MIA| N019AA|             Empty|      Empty|            Empty|           Empty|     Empty|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|             Empty|      Empty|            Empty|           Empty|     Empty|\n",
      "|    16.0|2015-01-01 15:30:00|2015-01-01 14:25:00|     AA|         1|        4|        1|    24.0| DFW|   190.0|2015-01-01|     2297|   AUS|AUS-DFW| N483AA|             Empty|      Empty|            Empty|           Empty|     Empty|\n",
      "|    25.0|2015-01-01 21:10:00|2015-01-01 19:45:00|     AA|         1|        4|        1|    28.0| LAX|  1242.0|2015-01-01|     1167|   AUS|AUS-LAX| N4XVAA|             Empty|      Empty|            Empty|           Empty|     Empty|\n",
      "|    -6.0|2015-01-01 16:55:00|2015-01-01 14:20:00|     AA|         1|        4|        1|    13.0| ORD|   978.0|2015-01-01|     1201|   AUS|AUS-ORD| N566AA|             P & W|JT8D SERIES|MCDONNELL DOUGLAS|            1987|     TEXAS|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType, DateType, TimestampType\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "schema = StructType([\n",
    "  StructField(\"ArrDelay\", DoubleType(), True),\n",
    "  StructField(\"CRSArrTime\", TimestampType(), True),\n",
    "  StructField(\"CRSDepTime\", TimestampType(), True),\n",
    "  StructField(\"Carrier\", StringType(), True),\n",
    "  StructField(\"DayOfMonth\", IntegerType(), True),\n",
    "  StructField(\"DayOfWeek\", IntegerType(), True),\n",
    "  StructField(\"DayOfYear\", IntegerType(), True),\n",
    "  StructField(\"DepDelay\", DoubleType(), True),\n",
    "  StructField(\"Dest\", StringType(), True),\n",
    "  StructField(\"Distance\", DoubleType(), True),\n",
    "  StructField(\"FlightDate\", DateType(), True),\n",
    "  StructField(\"FlightNum\", StringType(), True),\n",
    "  StructField(\"Origin\", StringType(), True),\n",
    "  StructField(\"Route\", StringType(), True),\n",
    "  StructField(\"TailNum\", StringType(), True),\n",
    "  StructField(\"EngineManufacturer\", StringType(), True),\n",
    "  StructField(\"EngineModel\", StringType(), True),\n",
    "  StructField(\"Manufacturer\", StringType(), True),\n",
    "  StructField(\"ManufacturerYear\", StringType(), True),\n",
    "  StructField(\"OwnerState\", StringType(), True),\n",
    "])\n",
    "\n",
    "input_path = \"../data/simple_flight_delay_features_airplanes.json\"\n",
    "features = spark.read.json(input_path, schema=schema)\n",
    "features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------------+-------------------+---------------+\n",
      "|         CRSDepTime|CRSDepHourOfDay|         CRSArrTime|CRSArrHourOfDay|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "|2015-01-01 07:00:00|              7|2015-01-01 08:52:00|              8|\n",
      "|2015-01-01 21:10:00|             21|2015-01-01 23:02:00|             23|\n",
      "|2015-01-01 14:25:00|             14|2015-01-01 15:30:00|             15|\n",
      "|2015-01-01 19:45:00|             19|2015-01-01 21:10:00|             21|\n",
      "|2015-01-01 14:20:00|             14|2015-01-01 16:55:00|             16|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Add the hour of day of scheduled arrival/departure\n",
    "#\n",
    "from pyspark.sql.functions import hour\n",
    "features_with_hour = features.withColumn(\n",
    "  \"CRSDepHourOfDay\",\n",
    "  hour(features.CRSDepTime)\n",
    ")\n",
    "features_with_hour = features_with_hour.withColumn(\n",
    "  \"CRSArrHourOfDay\",\n",
    "  hour(features.CRSArrTime)\n",
    ")\n",
    "features_with_hour.select(\"CRSDepTime\", \"CRSDepHourOfDay\", \"CRSArrTime\", \"CRSArrHourOfDay\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because we left joined our new features in, we need to know how many of the resulting training records have `null` values for their fields. Null values will crash the `StringIndexer` for a field, so we’ve explicitly altered our feature extraction code to remove them. There should be no nulls, so we’ll print a table with a warning if they are present:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Null Value Report\n",
      "-----------------\n",
      "Column    Nulls\n",
      "--------  -------\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Check for nulls in features before using Spark ML\n",
    "#\n",
    "null_counts = [(column, features_with_hour.where(features_with_hour[column].isNull()).count()) for column in features_with_hour.columns]\n",
    "cols_with_nulls = filter(lambda x: x[1] > 0, null_counts)\n",
    "print(\"\\nNull Value Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(cols_with_nulls, headers=[\"Column\", \"Nulls\"]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There should be no `nulls` present!\n",
    "\n",
    "Next we need to bucketize our data as per normal."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------+\n",
      "|ArrDelay|ArrDelayBucket|\n",
      "+--------+--------------+\n",
      "|   -10.0|           1.0|\n",
      "|    -3.0|           1.0|\n",
      "|    16.0|           2.0|\n",
      "|    25.0|           2.0|\n",
      "|    -6.0|           1.0|\n",
      "+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay into on-time, slightly late, very late (0, 1, 2)\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "# Setup the Bucketizer\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "arrival_bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "\n",
    "# Save the model\n",
    "arrival_bucketizer_path = \"../models/arrival_bucketizer_2.0.bin\"\n",
    "arrival_bucketizer.write().overwrite().save(arrival_bucketizer_path)\n",
    "\n",
    "# Apply the model\n",
    "ml_bucketized_features = arrival_bucketizer.transform(features_with_hour)\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we add the hour of day fields as normal, and we bucketize the `ArrDelay` field to get the `ArrDelayBucket`. Then we need to index all our string columns, including our new airplane features.\n",
    "\n",
    "Note that we are also making another change. We are moving the `DayOfMonth`, `DayOfWeek` and `DayOfYear` fields, along with `CRSDepHourOfDay` and `CRSArrHourOfDay` into numeric fields directly to be vectorized. This is because in thinking about it... indexing these numeric fields only adds a layer of abstraction, encoding them into numbers again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+------------------------+-----------------+------------------+----------------------+----------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|EngineManufacturer|EngineModel|     Manufacturer|ManufacturerYear|OwnerState|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|Carrier_index|Origin_index|Dest_index|Route_index|TailNum_index|EngineManufacturer_index|EngineModel_index|Manufacturer_index|ManufacturerYear_index|OwnerState_index|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+------------------------+-----------------+------------------+----------------------+----------------+\n",
      "|   -10.0|2015-01-01 08:52:00|2015-01-01 07:00:00|     AA|         1|        4|        1|    -4.0| MIA|   594.0|2015-01-01|       17|   ATL|ATL-MIA| N019AA|             Empty|      Empty|            Empty|           Empty|     Empty|              7|              8|           1.0|          4.0|         0.0|      22.0|      112.0|       1333.0|                     1.0|              1.0|               1.0|                   0.0|             2.0|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             21|             23|           1.0|          4.0|         0.0|      22.0|      112.0|       3032.0|                     1.0|              1.0|               1.0|                   0.0|             2.0|\n",
      "|    16.0|2015-01-01 15:30:00|2015-01-01 14:25:00|     AA|         1|        4|        1|    24.0| DFW|   190.0|2015-01-01|     2297|   AUS|AUS-DFW| N483AA|             Empty|      Empty|            Empty|           Empty|     Empty|             14|             15|           2.0|          4.0|        38.0|       1.0|      246.0|       1273.0|                     1.0|              1.0|               1.0|                   0.0|             2.0|\n",
      "|    25.0|2015-01-01 21:10:00|2015-01-01 19:45:00|     AA|         1|        4|        1|    28.0| LAX|  1242.0|2015-01-01|     1167|   AUS|AUS-LAX| N4XVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             19|             21|           2.0|          4.0|        38.0|       4.0|      559.0|       2431.0|                     1.0|              1.0|               1.0|                   0.0|             2.0|\n",
      "|    -6.0|2015-01-01 16:55:00|2015-01-01 14:20:00|     AA|         1|        4|        1|    13.0| ORD|   978.0|2015-01-01|     1201|   AUS|AUS-ORD| N566AA|             P & W|JT8D SERIES|MCDONNELL DOUGLAS|            1987|     TEXAS|             14|             16|           1.0|          4.0|        38.0|       2.0|     1292.0|       1358.0|                     5.0|              4.0|               6.0|                  29.0|             1.0|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+------------------------+-----------------+------------------+----------------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into indexes\n",
    "string_columns = [\"Carrier\", \"Origin\", \"Dest\", \"Route\",\n",
    "                  \"TailNum\", \"EngineManufacturer\", \"EngineModel\",\n",
    "                  \"Manufacturer\", \"ManufacturerYear\", \"OwnerState\"]\n",
    "for column in string_columns:\n",
    "  string_indexer = StringIndexer(\n",
    "    inputCol=column,\n",
    "    outputCol=column + \"_index\"\n",
    "  )\n",
    "  \n",
    "  string_indexer_model = string_indexer.fit(ml_bucketized_features)\n",
    "  ml_bucketized_features = string_indexer_model.transform(ml_bucketized_features)\n",
    "  \n",
    "  # Save the pipeline model\n",
    "  string_indexer_output_path = \"../models/string_indexer_model_4.0.{}.bin\".format(\n",
    "    column\n",
    "  )\n",
    "  string_indexer_model.write().overwrite().save(string_indexer_output_path)\n",
    "\n",
    "ml_bucketized_features.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we need to create a new `VectorAssembler` to combine our features into one feature vector, the column `Features_vec`. As before, an index field name is the field name with _index appended. This time around, we use a list comprehension to compute the index columns:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Carrier_index',\n",
       " 'Origin_index',\n",
       " 'Dest_index',\n",
       " 'Route_index',\n",
       " 'TailNum_index',\n",
       " 'EngineManufacturer_index',\n",
       " 'EngineModel_index',\n",
       " 'Manufacturer_index',\n",
       " 'ManufacturerYear_index',\n",
       " 'OwnerState_index']"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Combine continuous, numeric fields with indexes of nominal ones\n",
    "# ...into one feature vector\n",
    "numeric_columns = [\n",
    "  \"DepDelay\",\n",
    "  \"Distance\",\n",
    "  \"DayOfYear\",\n",
    "  \"DayOfMonth\",\n",
    "  \"DayOfWeek\",\n",
    "  \"CRSDepHourOfDay\",\n",
    "  \"CRSArrHourOfDay\"\n",
    "]\n",
    "index_columns = [column + \"_index\" for column in string_columns]\n",
    "index_columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|EngineManufacturer|EngineModel|     Manufacturer|ManufacturerYear|OwnerState|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|        Features_vec|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "|   -10.0|2015-01-01 08:52:00|2015-01-01 07:00:00|     AA|         1|        4|        1|    -4.0| MIA|   594.0|2015-01-01|       17|   ATL|ATL-MIA| N019AA|             Empty|      Empty|            Empty|           Empty|     Empty|              7|              8|           1.0|[-4.0,594.0,1.0,1...|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             21|             23|           1.0|[-7.0,594.0,1.0,1...|\n",
      "|    16.0|2015-01-01 15:30:00|2015-01-01 14:25:00|     AA|         1|        4|        1|    24.0| DFW|   190.0|2015-01-01|     2297|   AUS|AUS-DFW| N483AA|             Empty|      Empty|            Empty|           Empty|     Empty|             14|             15|           2.0|[24.0,190.0,1.0,1...|\n",
      "|    25.0|2015-01-01 21:10:00|2015-01-01 19:45:00|     AA|         1|        4|        1|    28.0| LAX|  1242.0|2015-01-01|     1167|   AUS|AUS-LAX| N4XVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             19|             21|           2.0|[28.0,1242.0,1.0,...|\n",
      "|    -6.0|2015-01-01 16:55:00|2015-01-01 14:20:00|     AA|         1|        4|        1|    13.0| ORD|   978.0|2015-01-01|     1201|   AUS|AUS-ORD| N566AA|             P & W|JT8D SERIES|MCDONNELL DOUGLAS|            1987|     TEXAS|             14|             16|           1.0|[13.0,978.0,1.0,1...|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vector_assembler = VectorAssembler(\n",
    "  inputCols=numeric_columns + index_columns,\n",
    "  outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Save the numeric vector assembler\n",
    "vector_assembler_path = \"../models/numeric_vector_assembler_5.0.bin\"\n",
    "vector_assembler.write().overwrite().save(vector_assembler_path)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "  final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Inspect the finalized features\n",
    "final_vectorized_features.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The rest of the code is identical to [ch09/improved_spark_mllib_model.py](improved_spark_mllib_model.py):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Run 1 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5799608355091384\n",
      "weightedPrecision = 0.5128851827971685\n",
      "weightedRecall = 0.5799608355091384\n",
      "f1 = 0.5129849127510245\n",
      "\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5870511425462459\n",
      "weightedPrecision = 0.5216138118522686\n",
      "weightedRecall = 0.5870511425462459\n",
      "f1 = 0.520345356453493\n",
      "\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5797181237612861\n",
      "weightedPrecision = 0.512201626392988\n",
      "weightedRecall = 0.5797181237612861\n",
      "f1 = 0.5142372027931253\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "feature_importances = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(f\"\\nRun {i} out of {split_count} of test/train splits in cross validation...\")\n",
    "  \n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "  \n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4896,\n",
    "    maxMemoryInMB=1024\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "  \n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "  \n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "  \n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "    \n",
    "    scores[metric_name].append(score)\n",
    "    print(\"{} = {}\".format(metric_name, score))\n",
    "  \n",
    "  #\n",
    "  # Collect feature importances\n",
    "  #\n",
    "  feature_names = vector_assembler.getInputCols()\n",
    "  feature_importance_list = model.featureImportances\n",
    "  for feature_name, feature_importance in zip(feature_names, feature_importance_list):\n",
    "    feature_importances[feature_name].append(feature_importance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Log\n",
      "--------------\n",
      "Metric               Average         STD\n",
      "-----------------  ---------  ----------\n",
      "accuracy            0.582243  0.00340105\n",
      "weightedPrecision   0.515567  0.00428493\n",
      "weightedRecall      0.582243  0.00340105\n",
      "f1                  0.515856  0.00321548\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Evaluate average and STD of each metric and print a table\n",
    "#\n",
    "import numpy as np\n",
    "score_averages = defaultdict(float)\n",
    "\n",
    "# Compute the table data\n",
    "average_stds = []  # ha\n",
    "for metric_name in metric_names:\n",
    "  metric_scores = scores[metric_name]\n",
    "  \n",
    "  average_accuracy = sum(metric_scores) / len(metric_scores)\n",
    "  score_averages[metric_name] = average_accuracy\n",
    "  \n",
    "  std_accuracy = np.std(metric_scores)\n",
    "  \n",
    "  average_stds.append((metric_name, average_accuracy, std_accuracy))\n",
    "\n",
    "# Print the table\n",
    "print(\"\\nExperiment Log\")\n",
    "print(\"--------------\")\n",
    "print(tabulate(average_stds, headers=[\"Metric\", \"Average\", \"STD\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Report\n",
      "-----------------\n",
      "Metric                   Score\n",
      "-----------------  -----------\n",
      "accuracy           -0.00428316\n",
      "weightedPrecision  -0.145773\n",
      "weightedRecall     -0.00428316\n",
      "f1                 -0.00692455\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Persist the score to a sccore log that exists between runs\n",
    "#\n",
    "import pickle\n",
    "\n",
    "# Load the score log or initialize an empty one\n",
    "try:\n",
    "  score_log_filename = \"../models/score_log.pickle\"\n",
    "  score_log = pickle.load(open(score_log_filename, \"rb\"))\n",
    "  if not isinstance(score_log, list):\n",
    "    score_log = []\n",
    "except IOError:\n",
    "  score_log = []\n",
    "\n",
    "# Compute the existing score log entry\n",
    "score_log_entry = {\n",
    "  metric_name: score_averages[metric_name] for metric_name in metric_names\n",
    "}\n",
    "\n",
    "# Compute and display the change in score for each metric\n",
    "try:\n",
    "  last_log = score_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_log = score_log_entry\n",
    "\n",
    "experiment_report = []\n",
    "for metric_name in metric_names:\n",
    "  run_delta = score_log_entry[metric_name] - last_log[metric_name]\n",
    "  experiment_report.append((metric_name, run_delta))\n",
    "\n",
    "print(\"\\nExperiment Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(experiment_report, headers=[\"Metric\", \"Score\"]))\n",
    "\n",
    "# Append the existing average scores to the log\n",
    "score_log.append(score_log_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(score_log, open(score_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importances\n",
      "-------------------\n",
      "Name                        Importance\n",
      "------------------------  ------------\n",
      "DepDelay                    0.818825\n",
      "DayOfYear                   0.0389761\n",
      "DayOfMonth                  0.0287786\n",
      "Origin_index                0.0245126\n",
      "TailNum_index               0.0232409\n",
      "Route_index                 0.0183321\n",
      "Distance                    0.0124914\n",
      "Carrier_index               0.00840604\n",
      "Dest_index                  0.00840502\n",
      "CRSDepHourOfDay             0.00524377\n",
      "CRSArrHourOfDay             0.00495677\n",
      "EngineModel_index           0.00227696\n",
      "Manufacturer_index          0.00150245\n",
      "ManufacturerYear_index      0.00136106\n",
      "OwnerState_index            0.00127585\n",
      "EngineManufacturer_index    0.00111694\n",
      "DayOfWeek                   0.00029805\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Analyze and report feature importance changes\n",
    "#\n",
    "\n",
    "# Compute averages for each feature\n",
    "feature_importance_entry = defaultdict(float)\n",
    "for feature_name, value_list in feature_importances.items():\n",
    "  average_importance = sum(value_list) / len(value_list)\n",
    "  feature_importance_entry[feature_name] = average_importance\n",
    "\n",
    "# Sort the feature importances in descending order and print\n",
    "import operator\n",
    "sorted_feature_importances = sorted(\n",
    "  feature_importance_entry.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "print(\"\\nFeature Importances\")\n",
    "print(\"-------------------\")\n",
    "print(tabulate(sorted_feature_importances, headers=['Name', 'Importance']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importance Delta Report\n",
      "-------------------------------\n",
      "Feature                          Delta\n",
      "------------------------  ------------\n",
      "DayOfYear                  0.0389761\n",
      "DayOfMonth                 0.0287786\n",
      "TailNum_index              0.0232409\n",
      "Origin_index               0.00718938\n",
      "CRSArrHourOfDay            0.00319651\n",
      "CRSDepHourOfDay            0.00264478\n",
      "EngineModel_index          0.00227696\n",
      "Carrier_index              0.0018959\n",
      "Manufacturer_index         0.00150245\n",
      "ManufacturerYear_index     0.00136106\n",
      "OwnerState_index           0.00127585\n",
      "EngineManufacturer_index   0.00111694\n",
      "Dest_index                 0.00106729\n",
      "Route_index                0.000551038\n",
      "DayOfWeek                  0.00029805\n",
      "Distance                   0.000239563\n",
      "DepDelay                  -0.0449692\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Compare this run's feature importances with the previous run's\n",
    "#\n",
    "\n",
    "# Load the feature importance log or initialize an empty one\n",
    "try:\n",
    "  feature_log_filename = \"../models/feature_log.pickle\"\n",
    "  feature_log = pickle.load(open(feature_log_filename, \"rb\"))\n",
    "  if not isinstance(feature_log, list):\n",
    "    feature_log = []\n",
    "except IOError:\n",
    "  feature_log = []\n",
    "\n",
    "# Compute and display the change in score for each feature\n",
    "try:\n",
    "  last_feature_log = feature_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_feature_log = defaultdict(float)\n",
    "  for feature_name, importance in feature_importance_entry.items():\n",
    "    last_feature_log[feature_name] = importance\n",
    "\n",
    "# Compute the deltas\n",
    "feature_deltas = {}\n",
    "for feature_name in feature_importances.keys():\n",
    "  run_delta = feature_importance_entry[feature_name] - last_feature_log[feature_name]\n",
    "  feature_deltas[feature_name] = run_delta\n",
    "\n",
    "# Sort feature deltas, biggest change first\n",
    "import operator\n",
    "sorted_feature_deltas = sorted(\n",
    "  feature_deltas.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "# Display sorted feature deltas\n",
    "print(\"\\nFeature Importance Delta Report\")\n",
    "print(\"-------------------------------\")\n",
    "print(tabulate(sorted_feature_deltas, headers=[\"Feature\", \"Delta\"]))\n",
    "\n",
    "# Append the existing average deltas to the log\n",
    "feature_log.append(feature_importance_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(feature_log, open(feature_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that on the first go around, our model failed because we needed to increase the maxBins parameter to 4896 to accommodate our new fields. After that, the code runs without incident.\n",
    "\n",
    "It looks like our efforts were mostly for naught—they actually hurt the quality of the model (although so little that it comes out about even)! The single exception is that adding the `TailNum` helps in terms of feature importance by 0.05 (your precise results may vary). Apparently some airplanes are more prone to delay than others, but this isn’t down to the properties of the airplane much... or at least those properties of the airplane are encoded mostly by the identity of the airplane itself.\n",
    "\n",
    "### Keep It Simple Stupid: KISS Your Models\n",
    "\n",
    "The tail end of the scores for feature importance look like this (for me):\n",
    "\n",
    "```\n",
    "EngineModel_index          0.00221025\n",
    "OwnerState_index           0.00181267\n",
    "ManufacturerYear_index     0.00156983\n",
    "Manufacturer_index         0.000969526\n",
    "EngineManufacturer_index   0.000708076\n",
    "DayOfWeek                  0.00032268\n",
    "```\n",
    "\n",
    "It is better to have a simpler model as they tend not to _overfit_, that is to work well on training data by extracting patterns that only apply to it by chance but not to work well on test data or in the real world. They are also easier to deploy and maintain. If something doesn't help - remove it!\n",
    "\n",
    "## Model Simplification Experiment\n",
    "\n",
    "Lets remove `EngineModel`, `OwnerState`, `ManufacturerYear`, `Manufacturer`, `EngineManufacturer` and `DayOfWeek` and run the model again, to see what this does to performance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------+\n",
      "|ArrDelay|ArrDelayBucket|\n",
      "+--------+--------------+\n",
      "|   -10.0|           1.0|\n",
      "|    -3.0|           1.0|\n",
      "|    16.0|           2.0|\n",
      "|    25.0|           2.0|\n",
      "|    -6.0|           1.0|\n",
      "+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay into on-time, slightly late, very late (0, 1, 2)\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "# Setup the Bucketizer\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "arrival_bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "\n",
    "# Save the model\n",
    "arrival_bucketizer_path = \"../models/arrival_bucketizer_2.0.bin\"\n",
    "arrival_bucketizer.write().overwrite().save(arrival_bucketizer_path)\n",
    "\n",
    "# Apply the model\n",
    "ml_bucketized_features = arrival_bucketizer.transform(features_with_hour)\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "ename": "IllegalArgumentException",
     "evalue": "'requirement failed: Output column Carrier_index already exists.'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[0;32m~/spark/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m     62\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 63\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0ma\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     64\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[0;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[1;32m    327\u001b[0m                     \u001b[0;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 328\u001b[0;31m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[1;32m    329\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mPy4JJavaError\u001b[0m: An error occurred while calling o7367.fit.\n: java.lang.IllegalArgumentException: requirement failed: Output column Carrier_index already exists.\n\tat scala.Predef$.require(Predef.scala:224)\n\tat org.apache.spark.ml.feature.StringIndexerBase$class.validateAndTransformSchema(StringIndexer.scala:91)\n\tat org.apache.spark.ml.feature.StringIndexer.validateAndTransformSchema(StringIndexer.scala:109)\n\tat org.apache.spark.ml.feature.StringIndexer.transformSchema(StringIndexer.scala:152)\n\tat org.apache.spark.ml.PipelineStage.transformSchema(Pipeline.scala:74)\n\tat org.apache.spark.ml.feature.StringIndexer.fit(StringIndexer.scala:135)\n\tat sun.reflect.GeneratedMethodAccessor126.invoke(Unknown Source)\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.lang.reflect.Method.invoke(Method.java:498)\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\n\tat py4j.Gateway.invoke(Gateway.java:282)\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\n\tat java.lang.Thread.run(Thread.java:748)\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[0;31mIllegalArgumentException\u001b[0m                  Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-65-0155a341776d>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     12\u001b[0m   )\n\u001b[1;32m     13\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 14\u001b[0;31m   \u001b[0mstring_indexer_model\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mstring_indexer\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mml_bucketized_features\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     15\u001b[0m   \u001b[0mml_bucketized_features\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mstring_indexer_model\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtransform\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mml_bucketized_features\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     16\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark/python/pyspark/ml/base.py\u001b[0m in \u001b[0;36mfit\u001b[0;34m(self, dataset, params)\u001b[0m\n\u001b[1;32m    130\u001b[0m                 \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcopy\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mparams\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_fit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdataset\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    131\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 132\u001b[0;31m                 \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_fit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdataset\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    133\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    134\u001b[0m             raise ValueError(\"Params must be either a param map or a list/tuple of param maps, \"\n",
      "\u001b[0;32m~/spark/python/pyspark/ml/wrapper.py\u001b[0m in \u001b[0;36m_fit\u001b[0;34m(self, dataset)\u001b[0m\n\u001b[1;32m    293\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    294\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_fit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdataset\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 295\u001b[0;31m         \u001b[0mjava_model\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_fit_java\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdataset\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    296\u001b[0m         \u001b[0mmodel\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_create_model\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mjava_model\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    297\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_copyValues\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmodel\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark/python/pyspark/ml/wrapper.py\u001b[0m in \u001b[0;36m_fit_java\u001b[0;34m(self, dataset)\u001b[0m\n\u001b[1;32m    290\u001b[0m         \"\"\"\n\u001b[1;32m    291\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_transfer_params_to_java\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 292\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_java_obj\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdataset\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    293\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    294\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_fit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdataset\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1255\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1257\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1258\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1259\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/spark/python/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m     77\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mQueryExecutionException\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0ms\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m': '\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m1\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     78\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'java.lang.IllegalArgumentException: '\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 79\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mIllegalArgumentException\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0ms\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m': '\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m1\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     80\u001b[0m             \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     81\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0mdeco\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mIllegalArgumentException\u001b[0m: 'requirement failed: Output column Carrier_index already exists.'"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into indexes\n",
    "string_columns = [\"Carrier\", \"Origin\", \"Dest\", \"Route\", \"TailNum\"]\n",
    "for column in string_columns:\n",
    "  string_indexer = StringIndexer(\n",
    "    inputCol=column,\n",
    "    outputCol=column + \"_index\"\n",
    "  )\n",
    "  \n",
    "  string_indexer_model = string_indexer.fit(ml_bucketized_features)\n",
    "  ml_bucketized_features = string_indexer_model.transform(ml_bucketized_features)\n",
    "  \n",
    "  # Save the pipeline model\n",
    "  string_indexer_output_path = \"../models/string_indexer_model_4.0.{}.bin\".format(\n",
    "    column\n",
    "  )\n",
    "  string_indexer_model.write().overwrite().save(string_indexer_output_path)\n",
    "\n",
    "ml_bucketized_features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Carrier_index', 'Origin_index', 'Dest_index', 'Route_index', 'TailNum_index']"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Combine continuous, numeric fields with indexes of nominal ones\n",
    "# ...into one feature vector\n",
    "numeric_columns = [\n",
    "  \"DepDelay\",\n",
    "  \"Distance\",\n",
    "  \"DayOfYear\",\n",
    "  \"DayOfMonth\",\n",
    "  \"CRSDepHourOfDay\",\n",
    "  \"CRSArrHourOfDay\"\n",
    "]\n",
    "index_columns = [column + \"_index\" for column in string_columns]\n",
    "index_columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|EngineManufacturer|EngineModel|     Manufacturer|ManufacturerYear|OwnerState|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|        Features_vec|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "|   -10.0|2015-01-01 08:52:00|2015-01-01 07:00:00|     AA|         1|        4|        1|    -4.0| MIA|   594.0|2015-01-01|       17|   ATL|ATL-MIA| N019AA|             Empty|      Empty|            Empty|           Empty|     Empty|              7|              8|           1.0|[-4.0,594.0,1.0,1...|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             21|             23|           1.0|[-7.0,594.0,1.0,1...|\n",
      "|    16.0|2015-01-01 15:30:00|2015-01-01 14:25:00|     AA|         1|        4|        1|    24.0| DFW|   190.0|2015-01-01|     2297|   AUS|AUS-DFW| N483AA|             Empty|      Empty|            Empty|           Empty|     Empty|             14|             15|           2.0|[24.0,190.0,1.0,1...|\n",
      "|    25.0|2015-01-01 21:10:00|2015-01-01 19:45:00|     AA|         1|        4|        1|    28.0| LAX|  1242.0|2015-01-01|     1167|   AUS|AUS-LAX| N4XVAA|             Empty|      Empty|            Empty|           Empty|     Empty|             19|             21|           2.0|[28.0,1242.0,1.0,...|\n",
      "|    -6.0|2015-01-01 16:55:00|2015-01-01 14:20:00|     AA|         1|        4|        1|    13.0| ORD|   978.0|2015-01-01|     1201|   AUS|AUS-ORD| N566AA|             P & W|JT8D SERIES|MCDONNELL DOUGLAS|            1987|     TEXAS|             14|             16|           1.0|[13.0,978.0,1.0,1...|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+------------------+-----------+-----------------+----------------+----------+---------------+---------------+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vector_assembler = VectorAssembler(\n",
    "  inputCols=numeric_columns + index_columns,\n",
    "  outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Save the numeric vector assembler\n",
    "vector_assembler_path = \"../models/numeric_vector_assembler_5.0.bin\"\n",
    "vector_assembler.write().overwrite().save(vector_assembler_path)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "  final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Inspect the finalized features\n",
    "final_vectorized_features.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Run 1 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5889755011135858\n",
      "weightedPrecision = 0.7012102804705427\n",
      "weightedRecall = 0.5889755011135858\n",
      "f1 = 0.5232180887199622\n",
      "\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.579766107678729\n",
      "weightedPrecision = 0.633427591129964\n",
      "weightedRecall = 0.579766107678729\n",
      "f1 = 0.5149846630682994\n",
      "\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5836094543848023\n",
      "weightedPrecision = 0.51552540471056\n",
      "weightedRecall = 0.5836094543848023\n",
      "f1 = 0.5160721718042296\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "feature_importances = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(f\"\\nRun {i} out of {split_count} of test/train splits in cross validation...\")\n",
    "  \n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "  \n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4896,\n",
    "    maxMemoryInMB=1024\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "  \n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "  \n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "  \n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "    \n",
    "    scores[metric_name].append(score)\n",
    "    print(\"{} = {}\".format(metric_name, score))\n",
    "  \n",
    "  #\n",
    "  # Collect feature importances\n",
    "  #\n",
    "  feature_names = vector_assembler.getInputCols()\n",
    "  feature_importance_list = model.featureImportances\n",
    "  for feature_name, feature_importance in zip(feature_names, feature_importance_list):\n",
    "    feature_importances[feature_name].append(feature_importance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Log\n",
      "--------------\n",
      "Metric               Average         STD\n",
      "-----------------  ---------  ----------\n",
      "accuracy            0.584117  0.00377681\n",
      "weightedPrecision   0.616721  0.0767205\n",
      "weightedRecall      0.584117  0.00377681\n",
      "f1                  0.518092  0.00365203\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Evaluate average and STD of each metric and print a table\n",
    "#\n",
    "import numpy as np\n",
    "score_averages = defaultdict(float)\n",
    "\n",
    "# Compute the table data\n",
    "average_stds = []  # ha\n",
    "for metric_name in metric_names:\n",
    "  metric_scores = scores[metric_name]\n",
    "  \n",
    "  average_accuracy = sum(metric_scores) / len(metric_scores)\n",
    "  score_averages[metric_name] = average_accuracy\n",
    "  \n",
    "  std_accuracy = np.std(metric_scores)\n",
    "  \n",
    "  average_stds.append((metric_name, average_accuracy, std_accuracy))\n",
    "\n",
    "# Print the table\n",
    "print(\"\\nExperiment Log\")\n",
    "print(\"--------------\")\n",
    "print(tabulate(average_stds, headers=[\"Metric\", \"Average\", \"STD\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Report\n",
      "-----------------\n",
      "Metric                  Score\n",
      "-----------------  ----------\n",
      "accuracy           0.00187365\n",
      "weightedPrecision  0.101154\n",
      "weightedRecall     0.00187365\n",
      "f1                 0.00223582\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Persist the score to a sccore log that exists between runs\n",
    "#\n",
    "import pickle\n",
    "\n",
    "# Load the score log or initialize an empty one\n",
    "try:\n",
    "  score_log_filename = \"../models/score_log.pickle\"\n",
    "  score_log = pickle.load(open(score_log_filename, \"rb\"))\n",
    "  if not isinstance(score_log, list):\n",
    "    score_log = []\n",
    "except IOError:\n",
    "  score_log = []\n",
    "\n",
    "# Compute the existing score log entry\n",
    "score_log_entry = {\n",
    "  metric_name: score_averages[metric_name] for metric_name in metric_names\n",
    "}\n",
    "\n",
    "# Compute and display the change in score for each metric\n",
    "try:\n",
    "  last_log = score_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_log = score_log_entry\n",
    "\n",
    "experiment_report = []\n",
    "for metric_name in metric_names:\n",
    "  run_delta = score_log_entry[metric_name] - last_log[metric_name]\n",
    "  experiment_report.append((metric_name, run_delta))\n",
    "\n",
    "print(\"\\nExperiment Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(experiment_report, headers=[\"Metric\", \"Score\"]))\n",
    "\n",
    "# Append the existing average scores to the log\n",
    "score_log.append(score_log_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(score_log, open(score_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importances\n",
      "-------------------\n",
      "Name               Importance\n",
      "---------------  ------------\n",
      "DepDelay           0.848534\n",
      "DayOfYear          0.0332143\n",
      "TailNum_index      0.0279824\n",
      "DayOfMonth         0.0208377\n",
      "Origin_index       0.0207811\n",
      "Route_index        0.0155598\n",
      "Distance           0.0132175\n",
      "Dest_index         0.00951608\n",
      "Carrier_index      0.00496423\n",
      "CRSDepHourOfDay    0.00380001\n",
      "CRSArrHourOfDay    0.00159252\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Analyze and report feature importance changes\n",
    "#\n",
    "\n",
    "# Compute averages for each feature\n",
    "feature_importance_entry = defaultdict(float)\n",
    "for feature_name, value_list in feature_importances.items():\n",
    "  average_importance = sum(value_list) / len(value_list)\n",
    "  feature_importance_entry[feature_name] = average_importance\n",
    "\n",
    "# Sort the feature importances in descending order and print\n",
    "import operator\n",
    "sorted_feature_importances = sorted(\n",
    "  feature_importance_entry.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "print(\"\\nFeature Importances\")\n",
    "print(\"-------------------\")\n",
    "print(tabulate(sorted_feature_importances, headers=['Name', 'Importance']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Feature Importance Delta Report\n",
      "-------------------------------\n",
      "Feature                 Delta\n",
      "---------------  ------------\n",
      "DepDelay          0.0297091\n",
      "TailNum_index     0.00474146\n",
      "Dest_index        0.00111106\n",
      "Distance          0.000726107\n",
      "CRSDepHourOfDay  -0.00144376\n",
      "Route_index      -0.00277233\n",
      "CRSArrHourOfDay  -0.00336425\n",
      "Carrier_index    -0.00344181\n",
      "Origin_index     -0.0037315\n",
      "DayOfYear        -0.0057618\n",
      "DayOfMonth       -0.00794093\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Compare this run's feature importances with the previous run's\n",
    "#\n",
    "\n",
    "# Load the feature importance log or initialize an empty one\n",
    "try:\n",
    "  feature_log_filename = \"../models/feature_log.pickle\"\n",
    "  feature_log = pickle.load(open(feature_log_filename, \"rb\"))\n",
    "  if not isinstance(feature_log, list):\n",
    "    feature_log = []\n",
    "except IOError:\n",
    "  feature_log = []\n",
    "\n",
    "# Compute and display the change in score for each feature\n",
    "try:\n",
    "  last_feature_log = feature_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_feature_log = defaultdict(float)\n",
    "  for feature_name, importance in feature_importance_entry.items():\n",
    "    last_feature_log[feature_name] = importance\n",
    "\n",
    "# Compute the deltas\n",
    "feature_deltas = {}\n",
    "for feature_name in feature_importances.keys():\n",
    "  run_delta = feature_importance_entry[feature_name] - last_feature_log[feature_name]\n",
    "  feature_deltas[feature_name] = run_delta\n",
    "\n",
    "# Sort feature deltas, biggest change first\n",
    "import operator\n",
    "sorted_feature_deltas = sorted(\n",
    "  feature_deltas.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "# Display sorted feature deltas\n",
    "print(\"\\nFeature Importance Delta Report\")\n",
    "print(\"-------------------------------\")\n",
    "print(tabulate(sorted_feature_deltas, headers=[\"Feature\", \"Delta\"]))\n",
    "\n",
    "# Append the existing average deltas to the log\n",
    "feature_log.append(feature_importance_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(feature_log, open(feature_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Interpreting Our Results\n",
    "\n",
    "This impacts the score in a positive way (for me), but not in a significant way: an improvement of 0.00031884 in accuracy. However, at this point all our features are contributing significantly to the model’s prediction quality, which is where we want to be:\n",
    "\n",
    "```\n",
    "Feature Importances\n",
    "-------------------\n",
    "Name               Importance\n",
    "---------------  ------------\n",
    "DepDelay           0.775767\n",
    "TailNum_index      0.0541045\n",
    "Route_index        0.0401366\n",
    "Origin_index       0.0290746\n",
    "DayOfMonth         0.0287668\n",
    "DayOfYear          0.0268546\n",
    "Distance           0.0165887\n",
    "Dest_index         0.0126576\n",
    "CRSDepHourOfDay    0.00680116\n",
    "Carrier_index      0.00542581\n",
    "CRSArrHourOfDay    0.00382289\n",
    "```\n",
    "\n",
    "Remember: when it comes to predictive models, _simpler is better_. If a feature doesn’t sizably influence prediction accuracy, remove it. The model’s quality will increase, it will perform faster in production, and you will have an easier time understanding the impact of additional features on the model. A simpler model will be less susceptible to bias.\n",
    "\n",
    "## Incorporating Flight Time\n",
    "\n",
    "One thing we haven’t considered yet is the flight time. We should be able to subtract the takeoff time from the landing time and get the duration of the flight. Since distance is a top-3 feature, and the hour of day matters, it seems like flight time might eke out a bit more prediction quality. Let’s try!\n",
    "\n",
    "In the book, we computed this field by converting our ISO datetimes into unix times (seconds since 1970) and subtracted the takeoff time from the landing time. This gave us flight time in seconds. However, it turns out there is a field called `AirTime` which is the minutes in the air. Lets try adding `AirTime` and see how things work. \n",
    "\n",
    "### Availability of Feature at Runtime\n",
    "\n",
    "It is easy to make the mistake of incorporating a feature into your model that you can't reliabily retrieve in realtime to make a prediction. Remember - if you can't get it in realtime when you make a prediction, you can't incorporate it into the model... not if your model is going to work in the real world. This is a big difference between data science in Kaffle competitions and data science in practice.\n",
    "\n",
    "`AirTime` __would__ be available at runtime, _if we compute an expected `AirTime` by subtracting the scheduled arrival time `CRSArrTime` from the schedule departure `CRSDepTime`_, after converting both to unix time. We would then have to divide by 60 to get the expected `AirTime`. Would that work alright? One must reason about features... and in this case it seems like a valid feature, and a similar feature, `Distance` has a 1.65% relative feature importance (for me).\n",
    "\n",
    "Check out [ch09/extract_features_with_flight_time.py](extract_features_with_flight_time.py), which we copied from [ch09/extract_features_with_airplanes.py](extract_features_with_airplanes.py). We only need to change one line, our selectExpr, to add the date math for our FlightTime field:\n",
    "\n",
    "Lets add `AirTime` to our model and see whether it helps or not. We'll have to go back and load the `AirTime` column, then proceed with our experiment. We're going to put all the code in one block this time, to give you a code block you can use to add your own features to the model later.\n",
    "\n",
    "We simply add the `AirTime` column to our initial `SELECT` statement to Spark SQL. Note that sometimes `AirTime` is not present, resulting in null values. To address this, we will use Spark SQL's `COALESCE` function to impute (fill in) a `0.0` value in place of null. This way our model can use this missing data as evidence of timeliness, along with the values which are present."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+-------+-------+\n",
      "|FlightNum|FlightDate|DayOfWeek|DayOfMonth|DayOfYear|Carrier|Origin|Dest|Distance|DepDelay|ArrDelay|CRSDepTime|CRSArrTime|  Route|TailNum|AirTime|\n",
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+-------+-------+\n",
      "|     2349|2015-01-01|        4|         1|      1-1|     AA|   ORD| DFW|   802.0|     0.0|    26.0|      1845|      2115|ORD-DFW| N002AA|  129.0|\n",
      "|     1422|2015-01-01|        4|         1|      1-1|     AA|   DFW| HDN|   769.0|    78.0|    78.0|      0800|      0925|DFW-HDN| N003AA|  111.0|\n",
      "|      356|2015-01-01|        4|         1|      1-1|     AA|   ATL| DFW|   731.0|    -5.0|     1.0|      1640|      1805|ATL-DFW| N006AA|  122.0|\n",
      "|     2360|2015-01-01|        4|         1|      1-1|     AA|   GUC| DFW|   678.0|   199.0|   222.0|      1450|      1750|GUC-DFW| N012AA|   95.0|\n",
      "|      903|2015-01-01|        4|         1|      1-1|     AA|   LAX| EGE|   748.0|    -7.0|    16.0|      0830|      1145|LAX-EGE| N017AA|  112.0|\n",
      "|     1162|2015-01-01|        4|         1|      1-1|     AA|   DFW| JAC|  1047.0|    45.0|    74.0|      1740|      1935|DFW-JAC| N018AA|  148.0|\n",
      "|     1137|2015-01-01|        4|         1|      1-1|     AA|   IAD| MIA|   921.0|    -2.0|     7.0|      1721|      2010|IAD-MIA| N022AA|  133.0|\n",
      "|     1697|2015-01-01|        4|         1|      1-1|     AA|   MIA| STL|  1068.0|   101.0|    97.0|      1400|      1555|MIA-STL| N023AA|  150.0|\n",
      "|     2330|2015-01-01|        4|         1|      1-1|     AA|   MIA| IAD|   921.0|    82.0|    72.0|      2125|      2354|MIA-IAD| N023AA|  118.0|\n",
      "|     2321|2015-01-01|        4|         1|      1-1|     AA|   ORD| DFW|   802.0|     3.0|    19.0|      1220|      1455|ORD-DFW| N200AA|  135.0|\n",
      "+---------+----------+---------+----------+---------+-------+------+----+--------+--------+--------+----------+----------+-------+-------+-------+\n",
      "only showing top 10 rows\n",
      "\n",
      "+----------+---------------+----------+---------------+\n",
      "|CRSDepTime|CRSDepHourOfDay|CRSArrTime|CRSArrHourOfDay|\n",
      "+----------+---------------+----------+---------------+\n",
      "|      1845|              0|      2115|              0|\n",
      "|      0800|              0|      0925|              0|\n",
      "|      1640|              0|      1805|              0|\n",
      "|      1450|              0|      1750|              0|\n",
      "|      0830|              0|      1145|              0|\n",
      "+----------+---------------+----------+---------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-------+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+-------------------+---------+------+-------+-------+\n",
      "|AirTime|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|         FlightDate|FlightNum|Origin|  Route|TailNum|\n",
      "+-------+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+-------------------+---------+------+-------+-------+\n",
      "|  122.0|     1.0|2015-01-01 18:05:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    -5.0| DFW|   731.0|2015-01-01 00:00:00|      356|   ATL|ATL-DFW| N006AA|\n",
      "|   86.0|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01 00:00:00|      349|   ATL|ATL-MIA| N3KVAA|\n",
      "|   36.0|   -16.0|2015-01-01 13:05:00|2015-01-01 12:00:00|     AA|         1|        4|        1|    -7.0| DFW|   190.0|2015-01-01 00:00:00|     2230|   AUS|AUS-DFW| N574AA|\n",
      "|  164.0|    20.0|2015-01-01 10:06:00|2015-01-01 06:40:00|     AA|         1|        4|        1|    -6.0| MIA|  1194.0|2015-01-01 00:00:00|     2476|   BDL|BDL-MIA| N3LEAA|\n",
      "|  112.0|    26.0|2015-01-01 19:20:00|2015-01-01 17:10:00|     AA|         1|        4|        1|    -4.0| DFW|   597.0|2015-01-01 00:00:00|     1224|   BHM|BHM-DFW| N4WMAA|\n",
      "|  255.0|    -8.0|2015-01-01 08:55:00|2015-01-01 06:10:00|     AA|         1|        4|        1|    -3.0| LAX|  1797.0|2015-01-01 00:00:00|     1361|   BNA|BNA-LAX| N3KYAA|\n",
      "|  264.0|    10.0|2015-01-01 19:40:00|2015-01-01 16:05:00|     AA|         1|        4|        1|    -6.0| DFW|  1562.0|2015-01-01 00:00:00|     1448|   BOS|BOS-DFW| N3EGAA|\n",
      "|  381.0|     7.0|2015-01-01 12:35:00|2015-01-01 09:00:00|     AA|         1|        4|        1|    -3.0| LAX|  2611.0|2015-01-01 00:00:00|       25|   BOS|BOS-LAX| N3AVAA|\n",
      "|  131.0|    -9.0|2015-01-01 20:12:00|2015-01-01 17:26:00|     AA|         1|        4|        1|   -11.0| MIA|   946.0|2015-01-01 00:00:00|      357|   BWI|BWI-MIA| N3HAAA|\n",
      "|  152.0|   -14.0|2015-01-01 09:10:00|2015-01-01 07:05:00|     AA|         1|        4|        1|    -7.0| DFW|   926.0|2015-01-01 00:00:00|     2434|   CMH|CMH-DFW| N575AA|\n",
      "+-------+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+-------------------+---------+------+-------+-------+\n",
      "only showing top 10 rows\n",
      "\n",
      "Features with AirTime prepared!\n"
     ]
    }
   ],
   "source": [
    "# Load the on-time parquet file\n",
    "input_path = \"../data/january_performance.parquet\"\n",
    "on_time_dataframe = spark.read.parquet(input_path)\n",
    "on_time_dataframe.registerTempTable(\"on_time_performance\")\n",
    "\n",
    "# Filtering on FlightDate > 2/1 and a 10% sample are for a training course. \n",
    "# Feel free to use all the data if you have the RAM and the time!\n",
    "features = spark.sql(\"\"\"\n",
    "SELECT\n",
    "    FlightNum,\n",
    "    FlightDate,\n",
    "    DayOfWeek,\n",
    "    DayofMonth AS DayOfMonth,\n",
    "    CONCAT(Month, '-',  DayofMonth) AS DayOfYear,\n",
    "    Carrier,\n",
    "    Origin,\n",
    "    Dest,\n",
    "    Distance,\n",
    "    DepDelay,\n",
    "    ArrDelay,\n",
    "    CRSDepTime,\n",
    "    CRSArrTime,\n",
    "    CONCAT(Origin, '-', Dest) AS Route,\n",
    "    TailNum,\n",
    "    COALESCE(AirTime, 0.0) AS AirTime\n",
    "FROM on_time_performance\n",
    "WHERE FlightDate < '2015-02-01'\n",
    "\"\"\")\n",
    "features = features.sample(False, 0.1) \n",
    "\n",
    "# Filter nulls, they can't help us\n",
    "features = features.filter(\n",
    "    (features.ArrDelay.isNotNull())\n",
    "    &\n",
    "    (features.DepDelay.isNotNull())\n",
    ")\n",
    "features.show(10)\n",
    "\n",
    "#\n",
    "# Add the hour of day of scheduled arrival/departure\n",
    "#\n",
    "from pyspark.sql.functions import hour\n",
    "features_with_hour = features.withColumn(\n",
    "  \"CRSDepHourOfDay\",\n",
    "  hour(features.CRSDepTime)\n",
    ")\n",
    "features_with_hour = features_with_hour.withColumn(\n",
    "  \"CRSArrHourOfDay\",\n",
    "  hour(features.CRSArrTime)\n",
    ")\n",
    "features_with_hour.select(\"CRSDepTime\", \"CRSDepHourOfDay\", \"CRSArrTime\", \"CRSArrHourOfDay\").show(5)\n",
    "\n",
    "# We need to turn timestamps into timestamps, and not strings or numbers\n",
    "def convert_hours(hours_minutes):\n",
    "  hours = hours_minutes[:-2]\n",
    "  minutes = hours_minutes[-2:]\n",
    "  \n",
    "  if hours == '24':\n",
    "    hours = '23'\n",
    "    minutes = '59'\n",
    "  \n",
    "  time_string = \"{}:{}:00Z\".format(hours, minutes)\n",
    "  return time_string\n",
    "\n",
    "def compose_datetime(iso_date, time_string):\n",
    "  return \"{} {}\".format(iso_date, time_string)\n",
    "\n",
    "def create_iso_string(iso_date, hours_minutes):\n",
    "  time_string = convert_hours(hours_minutes)\n",
    "  full_datetime = compose_datetime(iso_date, time_string)\n",
    "  return full_datetime\n",
    "\n",
    "def create_datetime(iso_string):\n",
    "  return iso8601.parse_date(iso_string)\n",
    "\n",
    "def convert_datetime(iso_date, hours_minutes):\n",
    "  iso_string = create_iso_string(iso_date, hours_minutes)\n",
    "  dt = create_datetime(iso_string)\n",
    "  return dt\n",
    "\n",
    "def day_of_year(iso_date_string):\n",
    "  dt = iso8601.parse_date(iso_date_string)\n",
    "  doy = dt.timetuple().tm_yday\n",
    "  return doy\n",
    "\n",
    "def alter_feature_datetimes(row):\n",
    "  \n",
    "  flight_date = iso8601.parse_date(row['FlightDate'])\n",
    "  scheduled_dep_time = convert_datetime(row['FlightDate'], row['CRSDepTime'])\n",
    "  scheduled_arr_time = convert_datetime(row['FlightDate'], row['CRSArrTime'])\n",
    "  \n",
    "  # Handle overnight flights\n",
    "  if scheduled_arr_time < scheduled_dep_time:\n",
    "    scheduled_arr_time += datetime.timedelta(days=1)\n",
    "  \n",
    "  doy = day_of_year(row['FlightDate'])\n",
    "  \n",
    "  return {\n",
    "    'FlightNum': row['FlightNum'],\n",
    "    'FlightDate': flight_date,\n",
    "    'DayOfWeek': int(row['DayOfWeek']),\n",
    "    'DayOfMonth': int(row['DayOfMonth']),\n",
    "    'DayOfYear': doy,\n",
    "    'Carrier': row['Carrier'],\n",
    "    'Origin': row['Origin'],\n",
    "    'Dest': row['Dest'],\n",
    "    'Distance': row['Distance'],\n",
    "    'DepDelay': row['DepDelay'],\n",
    "    'ArrDelay': row['ArrDelay'],\n",
    "    'CRSDepTime': scheduled_dep_time,\n",
    "    'CRSArrTime': scheduled_arr_time,\n",
    "    'Route': row['Route'],\n",
    "    'TailNum': row['TailNum'],\n",
    "    'AirTime': row['AirTime']\n",
    "  }\n",
    "\n",
    "timestamp_features = features_with_hour.rdd.map(alter_feature_datetimes)\n",
    "timestamp_df = timestamp_features.toDF()\n",
    "\n",
    "# Explicitly sort the data and keep it sorted throughout. Leave nothing to chance.\n",
    "sorted_features = timestamp_df.sort(\n",
    "  timestamp_df.DayOfYear,\n",
    "  timestamp_df.Carrier,\n",
    "  timestamp_df.Origin,\n",
    "  timestamp_df.Dest,\n",
    "  timestamp_df.FlightNum,\n",
    "  timestamp_df.CRSDepTime,\n",
    "  timestamp_df.CRSArrTime,\n",
    ")\n",
    "\n",
    "sorted_features.show(10)\n",
    "\n",
    "# Store as a single json file\n",
    "output_path = \"../data/simple_flight_delay_features_flight_times.json\"\n",
    "sorted_features.repartition(1).write.mode(\"overwrite\").json(output_path)\n",
    "\n",
    "print(\"Features with AirTime prepared!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Training the Model\n",
    "\n",
    "Note that we still store our data to disk and then load it explicitly. This ensures its provenance and formatting are exactly as we expect, and have not been inferred... inference that could change later and throw off our model. Model imput must be as accurate and precise as possible!\n",
    "\n",
    "Here we just add `AirTime` to the columns we load, then we add it to the list of `numeric_columns`. The rest is the same as before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|AirTime|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+\n",
      "|     1.0|2015-01-01 18:05:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    -5.0| DFW|   731.0|2015-01-01|      356|   ATL|ATL-DFW| N006AA|  122.0|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|   86.0|\n",
      "|   -16.0|2015-01-01 13:05:00|2015-01-01 12:00:00|     AA|         1|        4|        1|    -7.0| DFW|   190.0|2015-01-01|     2230|   AUS|AUS-DFW| N574AA|   36.0|\n",
      "|    20.0|2015-01-01 10:06:00|2015-01-01 06:40:00|     AA|         1|        4|        1|    -6.0| MIA|  1194.0|2015-01-01|     2476|   BDL|BDL-MIA| N3LEAA|  164.0|\n",
      "|    26.0|2015-01-01 19:20:00|2015-01-01 17:10:00|     AA|         1|        4|        1|    -4.0| DFW|   597.0|2015-01-01|     1224|   BHM|BHM-DFW| N4WMAA|  112.0|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "|         CRSDepTime|CRSDepHourOfDay|         CRSArrTime|CRSArrHourOfDay|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "|2015-01-01 16:40:00|             16|2015-01-01 18:05:00|             18|\n",
      "|2015-01-01 21:10:00|             21|2015-01-01 23:02:00|             23|\n",
      "|2015-01-01 12:00:00|             12|2015-01-01 13:05:00|             13|\n",
      "|2015-01-01 06:40:00|              6|2015-01-01 10:06:00|             10|\n",
      "|2015-01-01 17:10:00|             17|2015-01-01 19:20:00|             19|\n",
      "+-------------------+---------------+-------------------+---------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      "Null Value Report\n",
      "-----------------\n",
      "Column    Nulls\n",
      "--------  -------\n",
      "+--------+--------------+\n",
      "|ArrDelay|ArrDelayBucket|\n",
      "+--------+--------------+\n",
      "|     1.0|           2.0|\n",
      "|    -3.0|           1.0|\n",
      "|   -16.0|           0.0|\n",
      "|    20.0|           2.0|\n",
      "|    26.0|           2.0|\n",
      "+--------+--------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|AirTime|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+\n",
      "|     1.0|2015-01-01 18:05:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    -5.0| DFW|   731.0|2015-01-01|      356|   ATL|ATL-DFW| N006AA|  122.0|             16|             18|           2.0|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|   86.0|             21|             23|           1.0|\n",
      "|   -16.0|2015-01-01 13:05:00|2015-01-01 12:00:00|     AA|         1|        4|        1|    -7.0| DFW|   190.0|2015-01-01|     2230|   AUS|AUS-DFW| N574AA|   36.0|             12|             13|           0.0|\n",
      "|    20.0|2015-01-01 10:06:00|2015-01-01 06:40:00|     AA|         1|        4|        1|    -6.0| MIA|  1194.0|2015-01-01|     2476|   BDL|BDL-MIA| N3LEAA|  164.0|              6|             10|           2.0|\n",
      "|    26.0|2015-01-01 19:20:00|2015-01-01 17:10:00|     AA|         1|        4|        1|    -4.0| DFW|   597.0|2015-01-01|     1224|   BHM|BHM-DFW| N4WMAA|  112.0|             17|             19|           2.0|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|AirTime|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|Carrier_index|Origin_index|Dest_index|Route_index|TailNum_index|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+\n",
      "|     1.0|2015-01-01 18:05:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    -5.0| DFW|   731.0|2015-01-01|      356|   ATL|ATL-DFW| N006AA|  122.0|             16|             18|           2.0|          4.0|         0.0|       1.0|       14.0|       2720.0|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|   86.0|             21|             23|           1.0|          4.0|         0.0|      21.0|      119.0|       1075.0|\n",
      "|   -16.0|2015-01-01 13:05:00|2015-01-01 12:00:00|     AA|         1|        4|        1|    -7.0| DFW|   190.0|2015-01-01|     2230|   AUS|AUS-DFW| N574AA|   36.0|             12|             13|           0.0|          4.0|        33.0|       1.0|      100.0|       2012.0|\n",
      "|    20.0|2015-01-01 10:06:00|2015-01-01 06:40:00|     AA|         1|        4|        1|    -6.0| MIA|  1194.0|2015-01-01|     2476|   BDL|BDL-MIA| N3LEAA|  164.0|              6|             10|           2.0|          4.0|        55.0|      21.0|     3462.0|       2060.0|\n",
      "|    26.0|2015-01-01 19:20:00|2015-01-01 17:10:00|     AA|         1|        4|        1|    -4.0| DFW|   597.0|2015-01-01|     1224|   BHM|BHM-DFW| N4WMAA|  112.0|             17|             19|           2.0|          4.0|        67.0|       1.0|     2846.0|       3703.0|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+-------------+------------+----------+-----------+-------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+--------------------+\n",
      "|ArrDelay|         CRSArrTime|         CRSDepTime|Carrier|DayOfMonth|DayOfWeek|DayOfYear|DepDelay|Dest|Distance|FlightDate|FlightNum|Origin|  Route|TailNum|AirTime|CRSDepHourOfDay|CRSArrHourOfDay|ArrDelayBucket|        Features_vec|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+--------------------+\n",
      "|     1.0|2015-01-01 18:05:00|2015-01-01 16:40:00|     AA|         1|        4|        1|    -5.0| DFW|   731.0|2015-01-01|      356|   ATL|ATL-DFW| N006AA|  122.0|             16|             18|           2.0|[-5.0,731.0,1.0,1...|\n",
      "|    -3.0|2015-01-01 23:02:00|2015-01-01 21:10:00|     AA|         1|        4|        1|    -7.0| MIA|   594.0|2015-01-01|      349|   ATL|ATL-MIA| N3KVAA|   86.0|             21|             23|           1.0|[-7.0,594.0,1.0,1...|\n",
      "|   -16.0|2015-01-01 13:05:00|2015-01-01 12:00:00|     AA|         1|        4|        1|    -7.0| DFW|   190.0|2015-01-01|     2230|   AUS|AUS-DFW| N574AA|   36.0|             12|             13|           0.0|[-7.0,190.0,1.0,1...|\n",
      "|    20.0|2015-01-01 10:06:00|2015-01-01 06:40:00|     AA|         1|        4|        1|    -6.0| MIA|  1194.0|2015-01-01|     2476|   BDL|BDL-MIA| N3LEAA|  164.0|              6|             10|           2.0|[-6.0,1194.0,1.0,...|\n",
      "|    26.0|2015-01-01 19:20:00|2015-01-01 17:10:00|     AA|         1|        4|        1|    -4.0| DFW|   597.0|2015-01-01|     1224|   BHM|BHM-DFW| N4WMAA|  112.0|             17|             19|           2.0|[-4.0,597.0,1.0,1...|\n",
      "+--------+-------------------+-------------------+-------+----------+---------+---------+--------+----+--------+----------+---------+------+-------+-------+-------+---------------+---------------+--------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "\n",
      "Run 1 out of 3 of test/train splits in cross validation...\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "accuracy = 0.5834063047285464\n",
      "weightedPrecision = 0.5168915983398141\n",
      "weightedRecall = 0.5834063047285464\n",
      "f1 = 0.5203090213059611\n",
      "\n",
      "Run 2 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5826564215148189\n",
      "weightedPrecision = 0.5163211636392262\n",
      "weightedRecall = 0.5826564215148189\n",
      "f1 = 0.5196737570088943\n",
      "\n",
      "Run 3 out of 3 of test/train splits in cross validation...\n",
      "accuracy = 0.5780996649010918\n",
      "weightedPrecision = 0.5100745548772154\n",
      "weightedRecall = 0.5780996649010918\n",
      "f1 = 0.5145229403998973\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType, DateType, TimestampType\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.functions import udf\n",
    "\n",
    "schema = StructType([\n",
    "  StructField(\"ArrDelay\", DoubleType(), True),\n",
    "  StructField(\"CRSArrTime\", TimestampType(), True),\n",
    "  StructField(\"CRSDepTime\", TimestampType(), True),\n",
    "  StructField(\"Carrier\", StringType(), True),\n",
    "  StructField(\"DayOfMonth\", IntegerType(), True),\n",
    "  StructField(\"DayOfWeek\", IntegerType(), True),\n",
    "  StructField(\"DayOfYear\", IntegerType(), True),\n",
    "  StructField(\"DepDelay\", DoubleType(), True),\n",
    "  StructField(\"Dest\", StringType(), True),\n",
    "  StructField(\"Distance\", DoubleType(), True),\n",
    "  StructField(\"FlightDate\", DateType(), True),\n",
    "  StructField(\"FlightNum\", StringType(), True),\n",
    "  StructField(\"Origin\", StringType(), True),\n",
    "  StructField(\"Route\", StringType(), True),\n",
    "  StructField(\"TailNum\", StringType(), True),\n",
    "  StructField(\"AirTime\", FloatType(), True)\n",
    "])\n",
    "\n",
    "\n",
    "\n",
    "input_path = \"../data/simple_flight_delay_features_flight_times.json\"\n",
    "features = spark.read.json(input_path, schema=schema)\n",
    "features.show(5)\n",
    "\n",
    "#\n",
    "# Add the hour of day of scheduled arrival/departure\n",
    "#\n",
    "from pyspark.sql.functions import hour\n",
    "features_with_hour = features.withColumn(\n",
    "  \"CRSDepHourOfDay\",\n",
    "  hour(features.CRSDepTime)\n",
    ")\n",
    "features_with_hour = features_with_hour.withColumn(\n",
    "  \"CRSArrHourOfDay\",\n",
    "  hour(features.CRSArrTime)\n",
    ")\n",
    "features_with_hour.select(\"CRSDepTime\", \"CRSDepHourOfDay\", \"CRSArrTime\", \"CRSArrHourOfDay\").show(5)\n",
    "\n",
    "#\n",
    "# Check for nulls in features before using Spark ML\n",
    "#\n",
    "null_counts = [(column, features_with_hour.where(features_with_hour[column].isNull()).count()) for column in features_with_hour.columns]\n",
    "cols_with_nulls = filter(lambda x: x[1] > 0, null_counts)\n",
    "print(\"\\nNull Value Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(cols_with_nulls, headers=[\"Column\", \"Nulls\"]))\n",
    "\n",
    "#\n",
    "# Use pysmark.ml.feature.Bucketizer to bucketize ArrDelay into on-time, slightly late, very late (0, 1, 2)\n",
    "#\n",
    "from pyspark.ml.feature import Bucketizer\n",
    "\n",
    "# Setup the Bucketizer\n",
    "splits = [-float(\"inf\"), -15.0, 0, 30.0, float(\"inf\")]\n",
    "arrival_bucketizer = Bucketizer(\n",
    "  splits=splits,\n",
    "  inputCol=\"ArrDelay\",\n",
    "  outputCol=\"ArrDelayBucket\"\n",
    ")\n",
    "\n",
    "# Save the model\n",
    "arrival_bucketizer_path = \"../models/arrival_bucketizer_2.0.bin\"\n",
    "arrival_bucketizer.write().overwrite().save(arrival_bucketizer_path)\n",
    "\n",
    "# Apply the model\n",
    "ml_bucketized_features = arrival_bucketizer.transform(features_with_hour)\n",
    "ml_bucketized_features.select(\"ArrDelay\", \"ArrDelayBucket\").show(5)\n",
    "\n",
    "ml_bucketized_features.show(5)\n",
    "\n",
    "#\n",
    "# Extract features tools in with pyspark.ml.feature\n",
    "#\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler\n",
    "\n",
    "# Turn category fields into indexes\n",
    "string_columns = [\"Carrier\", \"Origin\", \"Dest\", \"Route\", \"TailNum\"]\n",
    "for column in string_columns:\n",
    "  string_indexer = StringIndexer(\n",
    "    inputCol=column,\n",
    "    outputCol=column + \"_index\"\n",
    "  )\n",
    "  \n",
    "  string_indexer_model = string_indexer.fit(ml_bucketized_features)\n",
    "  ml_bucketized_features = string_indexer_model.transform(ml_bucketized_features)\n",
    "  \n",
    "  # Save the pipeline model\n",
    "  string_indexer_output_path = \"../models/string_indexer_model_4.0.{}.bin\".format(\n",
    "    column\n",
    "  )\n",
    "  string_indexer_model.write().overwrite().save(string_indexer_output_path)\n",
    "\n",
    "ml_bucketized_features.show(5)\n",
    "\n",
    "# Combine continuous, numeric fields with indexes of nominal ones\n",
    "# ...into one feature vector\n",
    "numeric_columns = [\n",
    "  \"DepDelay\",\n",
    "  \"Distance\",\n",
    "  \"DayOfYear\",\n",
    "  \"DayOfMonth\",\n",
    "  \"CRSDepHourOfDay\",\n",
    "  \"CRSArrHourOfDay\",\n",
    "  \"AirTime\"\n",
    "]\n",
    "index_columns = [column + \"_index\" for column in string_columns]\n",
    "input_columns = numeric_columns + index_columns\n",
    "\n",
    "vector_assembler = VectorAssembler(\n",
    "  inputCols=input_columns,\n",
    "  outputCol=\"Features_vec\"\n",
    ")\n",
    "final_vectorized_features = vector_assembler.transform(ml_bucketized_features)\n",
    "\n",
    "# Save the numeric vector assembler\n",
    "vector_assembler_path = \"../models/numeric_vector_assembler_5.0.bin\"\n",
    "vector_assembler.write().overwrite().save(vector_assembler_path)\n",
    "\n",
    "# Drop the index columns\n",
    "for column in index_columns:\n",
    "  final_vectorized_features = final_vectorized_features.drop(column)\n",
    "\n",
    "# Inspect the finalized features\n",
    "final_vectorized_features.show(5)\n",
    "\n",
    "#\n",
    "# Cross validate, train and evaluate classifier: loop 5 times for 4 metrics\n",
    "#\n",
    "\n",
    "from collections import defaultdict\n",
    "scores = defaultdict(list)\n",
    "feature_importances = defaultdict(list)\n",
    "metric_names = [\"accuracy\", \"weightedPrecision\", \"weightedRecall\", \"f1\"]\n",
    "split_count = 3\n",
    "\n",
    "for i in range(1, split_count + 1):\n",
    "  print(\"\\nRun {} out of {} of test/train splits in cross validation...\".format(\n",
    "    i,\n",
    "    split_count,\n",
    "  )\n",
    "  )\n",
    "  \n",
    "  # Test/train split\n",
    "  training_data, test_data = final_vectorized_features.randomSplit([0.8, 0.2])\n",
    "  \n",
    "  # Instantiate and fit random forest classifier on all the data\n",
    "  from pyspark.ml.classification import RandomForestClassifier\n",
    "  rfc = RandomForestClassifier(\n",
    "    featuresCol=\"Features_vec\",\n",
    "    labelCol=\"ArrDelayBucket\",\n",
    "    predictionCol=\"Prediction\",\n",
    "    maxBins=4896,\n",
    "    maxMemoryInMB=1024\n",
    "  )\n",
    "  model = rfc.fit(training_data)\n",
    "  \n",
    "  # Save the new model over the old one\n",
    "  model_output_path = \"../models/spark_random_forest_classifier.flight_delays.baseline.bin\"\n",
    "  model.write().overwrite().save(model_output_path)\n",
    "  \n",
    "  # Evaluate model using test data\n",
    "  predictions = model.transform(test_data)\n",
    "  \n",
    "  # Evaluate this split's results for each metric\n",
    "  from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "  for metric_name in metric_names:\n",
    "    evaluator = MulticlassClassificationEvaluator(\n",
    "      labelCol=\"ArrDelayBucket\",\n",
    "      predictionCol=\"Prediction\",\n",
    "      metricName=metric_name\n",
    "    )\n",
    "    score = evaluator.evaluate(predictions)\n",
    "    \n",
    "    scores[metric_name].append(score)\n",
    "    print(\"{} = {}\".format(metric_name, score))\n",
    "  \n",
    "  #\n",
    "  # Collect feature importances\n",
    "  #\n",
    "  feature_names = vector_assembler.getInputCols()\n",
    "  feature_importance_list = model.featureImportances\n",
    "  for feature_name, feature_importance in zip(feature_names, feature_importance_list):\n",
    "    feature_importances[feature_name].append(feature_importance)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculating `AirTime` Performance\n",
    "\n",
    "Then we calculate performance again... this time we do everything in one block, again so you can easily copy/paste this below to add your own new features!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Experiment Log\n",
      "--------------\n",
      "Metric               Average         STD\n",
      "-----------------  ---------  ----------\n",
      "accuracy            0.581387  0.00234489\n",
      "weightedPrecision   0.514429  0.00308793\n",
      "weightedRecall      0.581387  0.00234489\n",
      "f1                  0.518169  0.00259086\n",
      "\n",
      "Experiment Report\n",
      "-----------------\n",
      "Metric                    Score\n",
      "-----------------  ------------\n",
      "accuracy           -0.00272956\n",
      "weightedPrecision  -0.102292\n",
      "weightedRecall     -0.00272956\n",
      "f1                  7.69317e-05\n",
      "\n",
      "Feature Importances\n",
      "-------------------\n",
      "Name               Importance\n",
      "---------------  ------------\n",
      "DepDelay           0.843461\n",
      "DayOfYear          0.0346129\n",
      "TailNum_index      0.0273644\n",
      "Route_index        0.0235947\n",
      "DayOfMonth         0.0233104\n",
      "Origin_index       0.0126936\n",
      "Distance           0.00883134\n",
      "Dest_index         0.00851761\n",
      "AirTime            0.00534284\n",
      "CRSArrHourOfDay    0.00514335\n",
      "Carrier_index      0.00464775\n",
      "CRSDepHourOfDay    0.00248028\n",
      "\n",
      "Feature Importance Delta Report\n",
      "-------------------------------\n",
      "Feature                 Delta\n",
      "---------------  ------------\n",
      "Route_index       0.00803488\n",
      "AirTime           0.00534284\n",
      "CRSArrHourOfDay   0.00355083\n",
      "DayOfMonth        0.00247274\n",
      "DayOfYear         0.00139867\n",
      "Carrier_index    -0.000316482\n",
      "TailNum_index    -0.000617967\n",
      "Dest_index       -0.00099847\n",
      "CRSDepHourOfDay  -0.00131973\n",
      "Distance         -0.00438617\n",
      "DepDelay         -0.00507362\n",
      "Origin_index     -0.00808751\n"
     ]
    }
   ],
   "source": [
    "#\n",
    "# Evaluate average and STD of each metric and print a table\n",
    "#\n",
    "import numpy as np\n",
    "score_averages = defaultdict(float)\n",
    "\n",
    "# Compute the table data\n",
    "average_stds = []  # ha\n",
    "for metric_name in metric_names:\n",
    "  metric_scores = scores[metric_name]\n",
    "  \n",
    "  average_accuracy = sum(metric_scores) / len(metric_scores)\n",
    "  score_averages[metric_name] = average_accuracy\n",
    "  \n",
    "  std_accuracy = np.std(metric_scores)\n",
    "  \n",
    "  average_stds.append((metric_name, average_accuracy, std_accuracy))\n",
    "\n",
    "# Print the table\n",
    "print(\"\\nExperiment Log\")\n",
    "print(\"--------------\")\n",
    "print(tabulate(average_stds, headers=[\"Metric\", \"Average\", \"STD\"]))\n",
    "\n",
    "#\n",
    "# Persist the score to a sccore log that exists between runs\n",
    "#\n",
    "import pickle\n",
    "\n",
    "# Load the score log or initialize an empty one\n",
    "try:\n",
    "  score_log_filename = \"../models/score_log.pickle\"\n",
    "  score_log = pickle.load(open(score_log_filename, \"rb\"))\n",
    "  if not isinstance(score_log, list):\n",
    "    score_log = []\n",
    "except IOError:\n",
    "  score_log = []\n",
    "\n",
    "# Compute the existing score log entry\n",
    "score_log_entry = {\n",
    "  metric_name: score_averages[metric_name] for metric_name in metric_names\n",
    "}\n",
    "\n",
    "# Compute and display the change in score for each metric\n",
    "try:\n",
    "  last_log = score_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_log = score_log_entry\n",
    "\n",
    "experiment_report = []\n",
    "for metric_name in metric_names:\n",
    "  run_delta = score_log_entry[metric_name] - last_log[metric_name]\n",
    "  experiment_report.append((metric_name, run_delta))\n",
    "\n",
    "print(\"\\nExperiment Report\")\n",
    "print(\"-----------------\")\n",
    "print(tabulate(experiment_report, headers=[\"Metric\", \"Score\"]))\n",
    "\n",
    "# Append the existing average scores to the log\n",
    "score_log.append(score_log_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(score_log, open(score_log_filename, \"wb\"))\n",
    "\n",
    "#\n",
    "# Analyze and report feature importance changes\n",
    "#\n",
    "\n",
    "# Compute averages for each feature\n",
    "feature_importance_entry = defaultdict(float)\n",
    "for feature_name, value_list in feature_importances.items():\n",
    "  average_importance = sum(value_list) / len(value_list)\n",
    "  feature_importance_entry[feature_name] = average_importance\n",
    "\n",
    "# Sort the feature importances in descending order and print\n",
    "import operator\n",
    "sorted_feature_importances = sorted(\n",
    "  feature_importance_entry.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "print(\"\\nFeature Importances\")\n",
    "print(\"-------------------\")\n",
    "print(tabulate(sorted_feature_importances, headers=['Name', 'Importance']))\n",
    "\n",
    "#\n",
    "# Compare this run's feature importances with the previous run's\n",
    "#\n",
    "\n",
    "# Load the feature importance log or initialize an empty one\n",
    "try:\n",
    "  feature_log_filename = \"../models/feature_log.pickle\"\n",
    "  feature_log = pickle.load(open(feature_log_filename, \"rb\"))\n",
    "  if not isinstance(feature_log, list):\n",
    "    feature_log = []\n",
    "except IOError:\n",
    "  feature_log = []\n",
    "\n",
    "# Compute and display the change in score for each feature\n",
    "try:\n",
    "  last_feature_log = feature_log[-1]\n",
    "except (IndexError, TypeError, AttributeError):\n",
    "  last_feature_log = defaultdict(float)\n",
    "  for feature_name, importance in feature_importance_entry.items():\n",
    "    last_feature_log[feature_name] = importance\n",
    "\n",
    "# Compute the deltas\n",
    "feature_deltas = {}\n",
    "for feature_name in feature_importances.keys():\n",
    "  run_delta = feature_importance_entry[feature_name] - last_feature_log[feature_name]\n",
    "  feature_deltas[feature_name] = run_delta\n",
    "\n",
    "# Sort feature deltas, biggest change first\n",
    "import operator\n",
    "sorted_feature_deltas = sorted(\n",
    "  feature_deltas.items(),\n",
    "  key=operator.itemgetter(1),\n",
    "  reverse=True\n",
    ")\n",
    "\n",
    "# Display sorted feature deltas\n",
    "print(\"\\nFeature Importance Delta Report\")\n",
    "print(\"-------------------------------\")\n",
    "print(tabulate(sorted_feature_deltas, headers=[\"Feature\", \"Delta\"]))\n",
    "\n",
    "# Append the existing average deltas to the log\n",
    "feature_log.append(feature_importance_entry)\n",
    "\n",
    "# Persist the log for next run\n",
    "pickle.dump(feature_log, open(feature_log_filename, \"wb\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Interpreting `AirTime` Performance\n",
    "\n",
    "This output suggests a significant improvement in performance! weightedPrecision is up by 0.14, and the FlightTime contributes about half a percent to the feature importance. Also note that the feature importance of FlightTime comes at the expense of Distance and DepDelay, which seems expected: Distance is conceptually similar to FlightTime, and DepDelay is the most important feature. Taken together, the performance and feature importance metrics indicate that FlightTime is a worthwhile improvement to our model:\n",
    "\n",
    "At this point, once again it seems that we’ve exhausted the possibilities of the date/time features (at least, without resorting to more sophisticated time series analysis techniques than I know).\n",
    "\n",
    "## Conclusion\n",
    "\n",
    "In this chapter we covered how to improve on our model using the data we’ve already collected. We can use this approach in combination with our ability to deploy applications to continuously improve our predictive systems."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "1. Look at the list of fields in the complete dataset at [https://www.transtats.bts.gov/Fields.asp](https://www.transtats.bts.gov/Fields.asp). Add another field to the model. It can be directly added or it can be derived from multiple values.\n",
    "\n",
    "2. Find and add an external feature to this model. Find another dataset which can be joined into this feature set to provide new features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
